Coverage for tests/test_reprocess_visit_image.py: 14%
219 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-09 04:36 -0700
1# This file is part of drp_tasks.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://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 program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <https://www.gnu.org/licenses/>.
22import tempfile
23import unittest
25import astropy.table
26import numpy as np
28import lsst.afw.image
29import lsst.afw.math
30import lsst.afw.table
31import lsst.daf.butler.tests as butlerTests
32import lsst.geom
33import lsst.meas.algorithms
34import lsst.meas.base.tests
35import lsst.pipe.base.testUtils
36import lsst.utils.tests
37from lsst.drp.tasks.reprocess_visit_image import ReprocessVisitImageTask
40def make_visit_summary(summary=None, psf=None, wcs=None, photo_calib=None, detector=42):
41 """Return a visit summary table with an entry for the given detector."""
42 if summary is None:
43 schema = lsst.afw.table.ExposureTable.makeMinimalSchema()
44 lsst.afw.image.ExposureSummaryStats.update_schema(schema)
45 summary = lsst.afw.table.ExposureCatalog(schema)
46 if summary.find(detector) is not None:
47 raise RuntimeError(f"Detector {detector} already exists in visit summary table, can't re-add it.")
49 record = summary.addNew()
50 record.setId(detector)
51 record.setApCorrMap(lsst.afw.image.ApCorrMap()),
53 record.setPsf(psf)
54 record.setPhotoCalib(photo_calib)
56 if wcs is not None:
57 record.setWcs(wcs)
58 else:
59 crpix = lsst.geom.Box2D(
60 lsst.geom.Box2D(lsst.geom.Point2D(0, 0), lsst.geom.Point2D(100, 100))
61 ).getCenter()
62 crval = lsst.geom.SpherePoint(45.0, 45.0, lsst.geom.degrees)
63 cdelt = 0.2 * lsst.geom.arcseconds
64 wcs = lsst.afw.geom.makeSkyWcs(
65 crpix=crpix, crval=crval, cdMatrix=lsst.afw.geom.makeCdMatrix(scale=cdelt)
66 )
67 record.setWcs(wcs)
69 lsst.afw.image.ExposureSummaryStats().update_record(record),
71 return summary
74class ReprocessVisitImageTaskTests(lsst.utils.tests.TestCase):
75 def setUp(self):
76 # Different x/y dimensions so they're easy to distinguish in a plot,
77 # and non-zero minimum, to help catch xy0 errors.
78 detector = 42
79 bbox = lsst.geom.Box2I(lsst.geom.Point2I(5, 4), lsst.geom.Point2I(205, 184))
80 self.sky_center = lsst.geom.SpherePoint(245.0, -45.0, lsst.geom.degrees)
81 self.photo_calib = 12.3
82 dataset = lsst.meas.base.tests.TestDataset(
83 bbox,
84 crval=self.sky_center,
85 calibration=self.photo_calib,
86 detector=detector,
87 # Force a large visitId, to test DM-49138.
88 visitId=2**33,
89 )
90 # sqrt of area of a normalized 2d gaussian
91 psf_scale = np.sqrt(4 * np.pi * (dataset.psfShape.getDeterminantRadius()) ** 2)
92 noise = 10.0 # stddev of noise per pixel
93 # Sources ordered from faintest to brightest.
94 self.fluxes = np.array(
95 (
96 6 * noise * psf_scale,
97 12 * noise * psf_scale,
98 45 * noise * psf_scale,
99 150 * noise * psf_scale,
100 400 * noise * psf_scale,
101 1000 * noise * psf_scale,
102 )
103 )
104 self.centroids = np.array(
105 ((162, 25), (40, 70), (100, 160), (50, 120), (92, 35), (175, 154)), dtype=np.float32
106 )
107 for flux, centroid in zip(self.fluxes, self.centroids):
108 dataset.addSource(instFlux=flux, centroid=lsst.geom.Point2D(centroid[0], centroid[1]))
110 # Bright extended source in the center of the image: should not appear
111 # in any of the output catalogs.
112 center = lsst.geom.Point2D(100, 100)
113 shape = lsst.afw.geom.Quadrupole(8, 9, 3)
114 dataset.addSource(instFlux=500 * noise * psf_scale, centroid=center, shape=shape)
116 schema = dataset.makeMinimalSchema()
117 self.truth_exposure, self.truth_cat = dataset.realize(noise=noise, schema=schema)
118 # Add a cosmic ray-like two hot pixels, to check CR removal.
119 self.truth_exposure.image[60, 60] = 10000
120 self.truth_exposure.image[60, 61] = 10000
122 # Make an exposure that looks like a PostISRCCD, to serve as the input.
123 self.exposure = lsst.afw.image.ExposureF()
124 self.exposure.maskedImage = self.truth_exposure.maskedImage.clone()
125 self.exposure.mask.clearMaskPlane(self.exposure.mask.getMaskPlane("DETECTED"))
126 self.exposure.mask.clearMaskPlane(self.exposure.mask.getMaskPlane("DETECTED_NEGATIVE"))
127 # PostISRCCD will have a VisitInfo and Detector attached.
128 self.exposure.info.setVisitInfo(self.truth_exposure.visitInfo)
129 self.exposure.info.setDetector(self.truth_exposure.getDetector())
131 # Subtract a background from the truth exposure, so that the input
132 # exposure still has the background in it.
133 config = lsst.meas.algorithms.SubtractBackgroundTask.ConfigClass()
134 # Don't really have a background, so have to fit simpler models.
135 config.approxOrderX = 1
136 task = lsst.meas.algorithms.SubtractBackgroundTask(config=config)
137 self.background = task.run(self.truth_exposure).background
138 self.visit_summary = make_visit_summary(
139 psf=self.truth_exposure.psf,
140 wcs=self.truth_exposure.wcs,
141 photo_calib=self.truth_exposure.photoCalib,
142 )
144 # A catalog that looks like the output of finalizeCharacterization,
145 # with a value set that we can test on the output.
146 self.visit_catalog = self.truth_cat.asAstropy()
147 self.visit_catalog.add_column(
148 astropy.table.Column(data=np.zeros(len(self.visit_catalog)), name="calib_psf_used", dtype=bool)
149 )
150 self.visit_catalog.add_column(
151 astropy.table.Column(
152 data=np.zeros(len(self.visit_catalog)), name="calib_psf_reserved", dtype=bool
153 )
154 )
155 self.visit_catalog.add_column(
156 astropy.table.Column(
157 data=np.zeros(len(self.visit_catalog)), name="calib_psf_candidate", dtype=bool
158 )
159 )
160 self.visit_catalog.add_column(
161 astropy.table.Column(data=[detector] * len(self.visit_catalog), name="detector")
162 )
163 # Marking faintest source, so it's easy to identify later.
164 self.visit_catalog["calib_psf_used"][0] = True
166 # Test-specific configuration:
167 self.config = ReprocessVisitImageTask.ConfigClass()
168 # Don't really have a background, so have to fit simpler models.
169 self.config.detection.background.approxOrderX = 1
170 # Only insert 2 sky sources, for simplicity.
171 self.config.sky_sources.nSources = 2
173 # Make a realistic id generator so that output catalog ids are useful.
174 # NOTE: The id generator is used to seed the noise replacer during
175 # measurement, so changes to values here can have subtle effects on
176 # the centroids and fluxes measured on the image, which might cause
177 # tests to fail.
178 data_id = lsst.daf.butler.DataCoordinate.standardize(
179 instrument="I",
180 visit=self.truth_exposure.visitInfo.id,
181 detector=12,
182 universe=lsst.daf.butler.DimensionUniverse(),
183 )
184 self.config.id_generator.packer.name = "observation"
185 # Without the LSSTCam-specific visitId handler, we have to use a large
186 # n_observation to fit visitId=2^33.
187 self.config.id_generator.packer["observation"].n_observations = 2**35
188 self.config.id_generator.packer["observation"].n_detectors = 99
189 self.config.id_generator.n_releases = 8
190 self.config.id_generator.release_id = 2
191 self.id_generator = self.config.id_generator.apply(data_id)
193 def test_run(self):
195 background_to_photometric_ratio_value = 1.1
197 for do_apply_flat_background_ratio in [False, True]:
198 config = self.config
199 config.do_apply_flat_background_ratio = do_apply_flat_background_ratio
201 if do_apply_flat_background_ratio:
202 background_to_photometric_ratio = self.exposure.image.clone()
203 background_to_photometric_ratio.array[:, :] = background_to_photometric_ratio_value
204 else:
205 background_to_photometric_ratio = None
207 task = ReprocessVisitImageTask(config=config)
208 result = task.run(
209 exposures=self.exposure.clone(),
210 initial_photo_calib=self.truth_exposure.photoCalib,
211 psf=self.truth_exposure.psf,
212 background=self.background,
213 ap_corr=lsst.afw.image.ApCorrMap(),
214 photo_calib=self.truth_exposure.photoCalib,
215 wcs=self.truth_exposure.wcs,
216 calib_sources=self.visit_catalog,
217 id_generator=self.id_generator,
218 background_to_photometric_ratio=background_to_photometric_ratio,
219 )
221 calibrated = result.exposure.photoCalib.calibrateImage(result.exposure.maskedImage)
222 self.assertImagesAlmostEqual(result.exposure.image, calibrated.image)
223 self.assertImagesAlmostEqual(result.exposure.variance, calibrated.variance)
224 self.assertEqual(result.exposure.psf, self.truth_exposure.psf)
225 self.assertEqual(result.exposure.wcs, self.truth_exposure.wcs)
226 # A calibrated exposure has PhotoCalib==1.
227 self.assertNotEqual(result.exposure.photoCalib, self.truth_exposure.photoCalib)
228 self.assertFloatsAlmostEqual(result.exposure.photoCalib.getCalibrationMean(), 1)
230 # All sources (plus sky sources) should have been detected.
231 self.assertEqual(len(result.sources), len(self.truth_cat) + self.config.sky_sources.nSources)
232 # Faintest non-sky source should be marked as used.
233 flux_sorted = result.sources[result.sources.argsort("slot_CalibFlux_instFlux")]
234 self.assertTrue(flux_sorted[~flux_sorted["sky_source"]]["calib_psf_used"][0])
235 # Test that the schema init-output agrees with the catalog output.
236 self.assertEqual(task.sources_schema.schema, result.sources_footprints.schema)
237 # The flux/instFlux ratio should be the LocalPhotoCalib value.
238 for record in result.sources_footprints:
239 self.assertAlmostEqual(
240 record["base_PsfFlux_flux"] / record["base_PsfFlux_instFlux"],
241 record["base_LocalPhotoCalib"],
242 )
244 if not do_apply_flat_background_ratio:
245 result_false = result
247 self.assertFloatsAlmostEqual(
248 result_false.sources_footprints["base_PsfFlux_instFlux"]
249 / result.sources_footprints["base_PsfFlux_instFlux"],
250 background_to_photometric_ratio_value,
251 rtol=1e-6,
252 )
254 self.assertFloatsAlmostEqual(
255 result_false.sources_footprints["base_PsfFlux_flux"]
256 / result.sources_footprints["base_PsfFlux_flux"],
257 background_to_photometric_ratio_value,
258 rtol=1e-6,
259 )
261 # Prior to DM-49138, LSSTCam-style >32-bit visitIds were silently
262 # down-cast to `0`.
263 self.assertTrue((result.sources["visit"] == 2**33).all())
265 self.assertEqual(result.exposure.metadata["BUNIT"], "nJy")
268class ReprocessVisitImageTaskRunQuantumTests(lsst.utils.tests.TestCase):
269 """Tests of ``ReprocessVisitImageTask.runQuantum``, which need a test
270 butler, but do not need real data.
271 """
273 def setUp(self):
274 instrument = "testCam"
275 exposure0 = 101
276 exposure1 = 102
277 visit = 100101
278 detector = 42
280 # Create a and populate a test butler for runQuantum tests.
281 self.repo_path = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
282 self.repo = butlerTests.makeTestRepo(self.repo_path.name)
284 # A complete instrument record is necessary for the id generator.
285 instrumentRecord = self.repo.dimensions["instrument"].RecordClass(
286 name=instrument,
287 visit_max=1e6,
288 exposure_max=1e6,
289 detector_max=128,
290 class_name="lsst.obs.base.instrument_tests.DummyCam",
291 )
292 self.repo.registry.syncDimensionData("instrument", instrumentRecord)
294 # dataIds for fake data
295 butlerTests.addDataIdValue(self.repo, "detector", detector)
296 butlerTests.addDataIdValue(self.repo, "detector", detector + 1)
297 butlerTests.addDataIdValue(self.repo, "detector", detector + 2)
298 butlerTests.addDataIdValue(self.repo, "exposure", exposure0)
299 butlerTests.addDataIdValue(self.repo, "exposure", exposure1)
300 butlerTests.addDataIdValue(self.repo, "visit", visit)
302 # inputs
303 butlerTests.addDatasetType(
304 self.repo, "postISRCCD", {"instrument", "exposure", "detector"}, "ExposureF"
305 )
306 butlerTests.addDatasetType(self.repo, "finalVisitSummary", {"instrument", "visit"}, "ExposureCatalog")
307 butlerTests.addDatasetType(
308 self.repo, "initial_pvi_background", {"instrument", "visit", "detector"}, "Background"
309 )
310 butlerTests.addDatasetType(
311 self.repo, "initial_photoCalib_detector", {"instrument", "visit", "detector"}, "PhotoCalib"
312 )
313 butlerTests.addDatasetType(self.repo, "skyCorr", {"instrument", "visit", "detector"}, "Background")
314 butlerTests.addDatasetType(self.repo, "finalized_src_table", {"instrument", "visit"}, "DataFrame")
315 butlerTests.addDatasetType(
316 self.repo, "background_to_photometric_ratio", {"instrument", "visit", "detector"}, "Image"
317 )
319 # outputs
320 butlerTests.addDatasetType(
321 self.repo, "source_schema", {"instrument", "visit", "detector"}, "SourceCatalog"
322 )
323 butlerTests.addDatasetType(self.repo, "pvi", {"instrument", "visit", "detector"}, "ExposureF")
324 butlerTests.addDatasetType(
325 self.repo, "sources_footprints_detector", {"instrument", "visit", "detector"}, "SourceCatalog"
326 )
327 butlerTests.addDatasetType(
328 self.repo, "sources_detector", {"instrument", "visit", "detector"}, "ArrowAstropy"
329 )
330 butlerTests.addDatasetType(
331 self.repo, "pvi_background", {"instrument", "visit", "detector"}, "Background"
332 )
334 # dataIds
335 self.exposure0_id = self.repo.registry.expandDataId(
336 {"instrument": instrument, "exposure": exposure0, "detector": detector}
337 )
338 self.exposure1_id = self.repo.registry.expandDataId(
339 {"instrument": instrument, "exposure": exposure1, "detector": detector}
340 )
341 self.visit_id = self.repo.registry.expandDataId(
342 {"instrument": instrument, "visit": visit, "detector": detector}
343 )
344 # Second id for testing on a detector that is not in visitSummary.
345 self.visit1_id = self.repo.registry.expandDataId(
346 {"instrument": instrument, "visit": visit, "detector": detector + 1}
347 )
348 # Third id for testing on a detector that is in visitSummary but
349 # has missing calibs.
350 self.visit2_id = self.repo.registry.expandDataId(
351 {"instrument": instrument, "visit": visit, "detector": detector + 2}
352 )
353 self.visit_only_id = self.repo.registry.expandDataId({"instrument": instrument, "visit": visit})
355 # put empty data
356 self.butler = butlerTests.makeTestCollection(self.repo)
357 self.butler.put(lsst.afw.image.ExposureF(), "postISRCCD", self.exposure0_id)
358 self.butler.put(lsst.afw.image.ExposureF(), "postISRCCD", self.exposure1_id)
359 control = lsst.afw.math.BackgroundControl(10, 10)
360 background = lsst.afw.math.makeBackground(lsst.afw.image.ImageF(100, 100), control)
361 self.butler.put(lsst.afw.image.PhotoCalib(10), "initial_photoCalib_detector", self.visit_id)
362 self.butler.put(lsst.afw.image.PhotoCalib(10), "initial_photoCalib_detector", self.visit1_id)
363 self.butler.put(lsst.afw.image.PhotoCalib(10), "initial_photoCalib_detector", self.visit2_id)
364 self.butler.put(lsst.afw.math.BackgroundList(background), "initial_pvi_background", self.visit_id)
365 self.butler.put(lsst.afw.math.BackgroundList(background), "initial_pvi_background", self.visit1_id)
366 self.butler.put(lsst.afw.math.BackgroundList(background), "initial_pvi_background", self.visit2_id)
367 self.butler.put(lsst.afw.math.BackgroundList(background), "skyCorr", self.visit_id)
368 self.butler.put(lsst.afw.math.BackgroundList(background), "skyCorr", self.visit1_id)
369 self.butler.put(lsst.afw.math.BackgroundList(background), "skyCorr", self.visit2_id)
370 self.butler.put(lsst.afw.table.SourceCatalog().asAstropy(), "finalized_src_table", self.visit_only_id)
371 self.butler.put(lsst.afw.image.ImageF(), "background_to_photometric_ratio", self.visit_id)
372 self.butler.put(lsst.afw.image.ImageF(), "background_to_photometric_ratio", self.visit1_id)
373 self.butler.put(lsst.afw.image.ImageF(), "background_to_photometric_ratio", self.visit2_id)
374 # Make a simple single gaussian psf so that psf is not None in
375 # finalVisitSummary table which would result in
376 # UpstreamFailureNoWorkFound being raised in ReprocessVisitImageTask,
377 # which will be independently tested with self.visit2_id.
378 simple_psf = lsst.meas.algorithms.SingleGaussianPsf(11, 11, 2.0)
379 photo_calib = lsst.afw.image.PhotoCalib(10, 0.5)
380 visit_summary = make_visit_summary(psf=simple_psf, photo_calib=photo_calib, detector=detector)
381 # Add a detector with an entry, but some required calibs set to None.
382 visit_summary = make_visit_summary(
383 summary=visit_summary, psf=None, photo_calib=None, detector=detector + 2
384 )
385 self.butler.put(visit_summary, "finalVisitSummary", self.visit_only_id)
387 def tearDown(self):
388 self.repo_path.cleanup()
390 def test_lintConnections(self):
391 """Check that the connections are self-consistent."""
392 Connections = ReprocessVisitImageTask.ConfigClass.ConnectionsClass
393 lsst.pipe.base.testUtils.lintConnections(Connections)
395 def test_runQuantum(self):
396 task = ReprocessVisitImageTask()
397 lsst.pipe.base.testUtils.assertValidInitOutput(task)
399 quantum = lsst.pipe.base.testUtils.makeQuantum(
400 task,
401 self.butler,
402 self.visit_id,
403 {
404 "exposures": [self.exposure0_id],
405 "visit_summary": self.visit_only_id,
406 "initial_photo_calib": self.visit_id,
407 "background_1": self.visit_id,
408 "background_2": self.visit_id,
409 "calib_sources": self.visit_only_id,
410 "background_to_photometric_ratio": self.visit_id,
411 # outputs
412 "exposure": self.visit_id,
413 "sources": self.visit_id,
414 "sources_footprints": self.visit_id,
415 "background": self.visit_id,
416 },
417 )
418 mock_run = lsst.pipe.base.testUtils.runTestQuantum(task, self.butler, quantum)
419 # Check that the proper kwargs are passed to run().
420 self.assertEqual(
421 mock_run.call_args.kwargs.keys(),
422 {
423 "exposures",
424 "initial_photo_calib",
425 "psf",
426 "background",
427 "ap_corr",
428 "photo_calib",
429 "wcs",
430 "calib_sources",
431 "id_generator",
432 "background_to_photometric_ratio",
433 "result",
434 },
435 )
437 def test_runQuantum_no_detector_in_visit_summary(self):
438 """Test how the task handles the detector not being in the input visit
439 summary.
440 """
441 task = ReprocessVisitImageTask()
442 lsst.pipe.base.testUtils.assertValidInitOutput(task)
444 quantum = lsst.pipe.base.testUtils.makeQuantum(
445 task,
446 self.butler,
447 self.visit1_id,
448 {
449 "exposures": [self.exposure0_id],
450 "visit_summary": self.visit_only_id,
451 "initial_photo_calib": self.visit1_id,
452 "background_1": self.visit1_id,
453 "background_2": self.visit1_id,
454 "calib_sources": self.visit_only_id,
455 "background_to_photometric_ratio": self.visit_id,
456 # outputs
457 "exposure": self.visit1_id,
458 "sources": self.visit1_id,
459 "sources_footprints": self.visit1_id,
460 "background": self.visit1_id,
461 },
462 )
463 msg = " > no entry for the detector was found in the visit summary table"
464 with self.assertRaisesRegex(
465 lsst.pipe.base.UpstreamFailureNoWorkFound, f"Skipping reprocessing of detector 43 because:\n{msg}"
466 ):
467 lsst.pipe.base.testUtils.runTestQuantum(task, self.butler, quantum)
469 def test_runQuantum_missing_calibs_for_detector_in_visit_summary(self):
470 """Test how the task handles the detector not being in the input visit
471 summary.
472 """
473 task = ReprocessVisitImageTask()
474 lsst.pipe.base.testUtils.assertValidInitOutput(task)
476 quantum = lsst.pipe.base.testUtils.makeQuantum(
477 task,
478 self.butler,
479 self.visit2_id,
480 {
481 "exposures": [self.exposure0_id],
482 "visit_summary": self.visit_only_id,
483 "initial_photo_calib": self.visit2_id,
484 "background_1": self.visit2_id,
485 "background_2": self.visit2_id,
486 "calib_sources": self.visit_only_id,
487 "background_to_photometric_ratio": self.visit_id,
488 # outputs
489 "exposure": self.visit2_id,
490 "sources": self.visit2_id,
491 "sources_footprints": self.visit2_id,
492 "background": self.visit2_id,
493 },
494 )
495 lines = [
496 " > the PSF model for the detector is None",
497 " > the photometric calibration model for the detector is None",
498 ]
499 msg = "\n".join(lines)
500 with self.assertRaisesRegex(
501 lsst.pipe.base.UpstreamFailureNoWorkFound, f"Skipping reprocessing of detector 44 because:\n{msg}"
502 ):
503 lsst.pipe.base.testUtils.runTestQuantum(task, self.butler, quantum)
505 def test_runQuantum_no_sky_corr(self):
506 """Test that the task will run if using the sky_corr input is
507 diabled.
508 """
509 config = ReprocessVisitImageTask.ConfigClass()
510 config.do_use_sky_corr = False
511 task = ReprocessVisitImageTask(config=config)
512 lsst.pipe.base.testUtils.assertValidInitOutput(task)
514 quantum = lsst.pipe.base.testUtils.makeQuantum(
515 task,
516 self.butler,
517 self.visit_id,
518 {
519 "exposures": [self.exposure0_id],
520 "visit_summary": self.visit_only_id,
521 "initial_photo_calib": self.visit_id,
522 "background_1": self.visit_id,
523 "calib_sources": self.visit_only_id,
524 "background_to_photometric_ratio": self.visit_id,
525 # outputs
526 "exposure": self.visit_id,
527 "sources": self.visit_id,
528 "sources_footprints": self.visit_id,
529 "background": self.visit_id,
530 },
531 )
532 mock_run = lsst.pipe.base.testUtils.runTestQuantum(task, self.butler, quantum)
533 # Check that the proper kwargs are passed to run().
534 self.assertEqual(
535 mock_run.call_args.kwargs.keys(),
536 {
537 "exposures",
538 "initial_photo_calib",
539 "psf",
540 "background",
541 "ap_corr",
542 "photo_calib",
543 "wcs",
544 "calib_sources",
545 "id_generator",
546 "background_to_photometric_ratio",
547 "result",
548 },
549 )
551 def test_runQuantum_illumination_correction(self):
552 """Test the task with illumination correction enabled."""
553 config = ReprocessVisitImageTask.ConfigClass()
554 config.do_apply_flat_background_ratio = True
555 task = ReprocessVisitImageTask(config=config)
556 lsst.pipe.base.testUtils.assertValidInitOutput(task)
558 quantum = lsst.pipe.base.testUtils.makeQuantum(
559 task,
560 self.butler,
561 self.visit_id,
562 {
563 "exposures": [self.exposure0_id],
564 "visit_summary": self.visit_only_id,
565 "initial_photo_calib": self.visit_id,
566 "background_1": self.visit_id,
567 "background_2": self.visit_id,
568 "calib_sources": self.visit_only_id,
569 "background_to_photometric_ratio": self.visit_id,
570 # outputs
571 "exposure": self.visit_id,
572 "sources": self.visit_id,
573 "sources_footprints": self.visit_id,
574 "background": self.visit_id,
575 },
576 )
577 mock_run = lsst.pipe.base.testUtils.runTestQuantum(task, self.butler, quantum)
578 # Check that the proper kwargs are passed to run().
579 self.assertEqual(
580 mock_run.call_args.kwargs.keys(),
581 {
582 "exposures",
583 "initial_photo_calib",
584 "psf",
585 "background",
586 "ap_corr",
587 "photo_calib",
588 "wcs",
589 "calib_sources",
590 "id_generator",
591 "background_to_photometric_ratio",
592 "result",
593 },
594 )
597def setup_module(module):
598 lsst.utils.tests.init()
601class MemoryTestCase(lsst.utils.tests.MemoryTestCase):
602 pass
605if __name__ == "__main__": 605 ↛ 606line 605 didn't jump to line 606 because the condition on line 605 was never true
606 lsst.utils.tests.init()
607 unittest.main()