Coverage for tests / test_driver.py: 36%

1# 

2# This file is part of ap_verify. 

3# 

4# Developed for the LSST Data Management System. 

5# This product includes software developed by the LSST Project 

6# (http://www.lsst.org). 

7# See the COPYRIGHT file at the top-level directory of this distribution 

8# for details of code ownership. 

9# 

10# This program is free software: you can redistribute it and/or modify 

11# it under the terms of the GNU General Public License as published by 

12# the Free Software Foundation, either version 3 of the License, or 

13# (at your option) any later version. 

14# 

15# This program is distributed in the hope that it will be useful, 

16# but WITHOUT ANY WARRANTY; without even the implied warranty of 

17# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

18# GNU General Public License for more details. 

19# 

20# You should have received a copy of the GNU General Public License 

21# along with this program. If not, see <http://www.gnu.org/licenses/>. 

22# 

23 

24import functools 

25import os 

26import shutil 

27import tempfile 

28import unittest.mock 

29 

30import lsst.utils.tests 

31from lsst.obs.base import RawIngestTask, DefineVisitsTask 

32from lsst.ap.verify import pipeline_driver 

33from lsst.ap.verify.testUtils import DataTestCase 

34from lsst.ap.verify import Dataset, WorkspaceGen3 

35 

36 

37def patchApPipeGen3(method): 

38 """Shortcut decorator for consistently patching AP code. 

39 """ 

40 @functools.wraps(method) 

41 def wrapper(self, *args, **kwargs): 

42 dbPatcher = unittest.mock.patch("lsst.ap.verify.pipeline_driver._makeApdb") 

43 pipePatcher = unittest.mock.patch("lsst.ap.verify.pipeline_driver.subprocess") 

44 patchedMethod = dbPatcher(pipePatcher(method)) 

45 return patchedMethod(self, *args, **kwargs) 

46 return wrapper 

47 

48 

49class PipelineDriverTestSuiteGen3(DataTestCase): 

50 

51 @classmethod 

52 def setUpClass(cls): 

53 super().setUpClass() 

54 

55 cls.dataset = Dataset(cls.testDataset) 

56 

57 def setUp(self): 

58 super().setUp() 

59 

60 self._testDir = tempfile.mkdtemp() 

61 self.addCleanup(shutil.rmtree, self._testDir, ignore_errors=True) 

62 

63 self.workspace = WorkspaceGen3(self._testDir) 

64 self.dataset.makeCompatibleRepoGen3(self.workspace.repo) 

65 raws = [os.path.join(self.dataset.rawLocation, "lsst_a_204595_R11_S01_i.fits")] 

66 rawIngest = RawIngestTask(butler=self.workspace.workButler, config=RawIngestTask.ConfigClass()) 

67 rawIngest.run(raws, run=None) 

68 defineVisit = DefineVisitsTask(butler=self.workspace.workButler, 

69 config=DefineVisitsTask.ConfigClass()) 

70 defineVisit.run(self.workspace.workButler.registry.queryDataIds("exposure", datasets="raw")) 

71 self.apPipeArgs = pipeline_driver.ApPipeParser().parse_args(["--pipeline", "foo.yaml"]) 

72 

73 def _getDbArgs(self, call_args): 

74 """Get the database args from the `_makeApdb` call signature. 

75 """ 

76 if call_args.args: 

77 return call_args.args[1] 

78 elif "args" in call_args.kwargs: 

79 return call_args.kwargs["args"] 

80 else: 

81 self.fail(f"No APDB args passed to {call_args}!") 

82 

83 def _getInstrument(self, call_args): 

84 """Get the instrument from the `_makeApdb` call signature. 

85 """ 

86 if "instrument" in call_args.kwargs: 

87 return call_args.kwargs["instrument"] 

88 elif call_args.args: 

89 return call_args.args[2] 

90 else: 

91 self.fail(f"No instrument passed to {call_args}!") 

92 

93 @patchApPipeGen3 

94 def testrunApPipeGen3WorkspaceDb(self, _mockPipe, mockDb): 

95 """Test that runApPipeGen3 places a database in the workspace location by default. 

96 """ 

97 pipeline_driver.runApPipeGen3(self.workspace, self.apPipeArgs) 

98 

99 mockDb.assert_called_once() 

100 dbArgs = self._getDbArgs(mockDb.call_args) 

101 self.assertIn("db_url", dbArgs) 

102 self.assertEqual(dbArgs["db_url"], "sqlite:///" + self.workspace.dbLocation) 

103 self.assertEqual(self._getInstrument(mockDb.call_args), "LSSTCam-imSim") 

104 

105 @patchApPipeGen3 

106 def testrunApPipeGen3WorkspaceCustom(self, _mockPipe, mockDb): 

107 """Test that runApPipeGen3 places a database in the specified location. 

108 """ 

109 self.apPipeArgs.db = "postgresql://somebody@pgdb.misc.org/custom_db" 

110 pipeline_driver.runApPipeGen3(self.workspace, self.apPipeArgs) 

111 

112 mockDb.assert_called_once() 

113 dbArgs = self._getDbArgs(mockDb.call_args) 

114 self.assertIn("db_url", dbArgs) 

115 self.assertEqual(dbArgs["db_url"], self.apPipeArgs.db) 

116 

117 @patchApPipeGen3 

118 def testrunApPipeGen3Reuse(self, mockPipe, _mockDb): 

119 """Test that runApPipeGen3 does not run the pipeline at all (not even with 

120 --skip-existing) if --skip-pipeline is provided. 

121 """ 

122 skipArgs = pipeline_driver.ApPipeParser().parse_args(["--skip-pipeline"]) 

123 pipeline_driver.runApPipeGen3(self.workspace, skipArgs) 

124 mockPipe.assert_not_called() 

125 

126 

127class MemoryTester(lsst.utils.tests.MemoryTestCase): 

128 pass 

129 

130 

131def setup_module(module): 

132 lsst.utils.tests.init() 

133 

134 

135if __name__ == "__main__": 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true

136 lsst.utils.tests.init() 

137 unittest.main()