Coverage for tests / test_cli_commands.py: 29%
70 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-01 08:22 +0000
1# This file is part of ctrl_bps.
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 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 <https://www.gnu.org/licenses/>.
27import unittest
29import click
31from lsst.ctrl.bps import BpsSubprocessError
32from lsst.ctrl.bps.cli import bps
33from lsst.ctrl.bps.cli.cmd.commands import catch_errors
34from lsst.daf.butler.cli.utils import LogCliRunner
37class TestCatchErrors(unittest.TestCase):
38 """Test context manager for catching BPS command errors."""
40 def testSuccess(self):
41 """Test a command that executes successfully."""
43 def testSubprocessFailure(self):
44 """Test if the error is handled when command's child process failed."""
46 def driver():
47 raise BpsSubprocessError(-9, "Subprocess failed")
49 with self.assertRaises(click.exceptions.Exit) as cm:
50 ctx = click.Context(click.Command("command"))
51 with ctx:
52 with catch_errors():
53 driver()
54 self.assertEqual(cm.exception.exit_code, -9)
56 def testFailure(self):
57 """Test if the error is handled when a command failed."""
59 def driver():
60 raise KeyError("Key not found.")
62 with self.assertRaisesRegex(KeyError, "Key not found"):
63 with catch_errors():
64 driver()
67class TestCommandPing(unittest.TestCase):
68 """Test executing the ping subcommand."""
70 def setUp(self):
71 self.runner = LogCliRunner()
73 def testPingNoArgs(self):
74 with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.ping_driver") as mock_driver:
75 mock_driver.return_value = 0
76 result = self.runner.invoke(bps.cli, ["ping"])
77 self.assertEqual(result.exit_code, 0)
78 mock_driver.assert_called_with(wms_service=None, pass_thru="")
80 def testPingClass(self):
81 with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.ping_driver") as mock_driver:
82 mock_driver.return_value = 0
83 result = self.runner.invoke(
84 bps.cli, ["ping", "--wms-service-class", "wms_test_utils.WmsServiceSuccess"]
85 )
86 self.assertEqual(result.exit_code, 0)
87 mock_driver.assert_called_with(wms_service="wms_test_utils.WmsServiceSuccess", pass_thru="")
89 def testPingFailure(self):
90 with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.ping_driver") as mock_driver:
91 mock_driver.return_value = 64 # avoid 1 and 2 as returned when cli problems
92 result = self.runner.invoke(
93 bps.cli, ["ping", "--wms-service-class", "wms_test_utils.WmsServiceFailure"]
94 )
95 self.assertEqual(result.exit_code, 64)
96 mock_driver.assert_called_with(
97 wms_service="wms_test_utils.WmsServiceFailure",
98 pass_thru="",
99 )
101 def testPingPassthru(self):
102 with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.ping_driver") as mock_driver:
103 mock_driver.return_value = 0
104 result = self.runner.invoke(
105 bps.cli,
106 [
107 "ping",
108 "--wms-service-class",
109 "wms_test_utils.WmsServicePassThru",
110 "--pass-thru",
111 "EXTRA_VALUES",
112 ],
113 )
114 self.assertEqual(result.exit_code, 0)
115 mock_driver.assert_called_with(
116 wms_service="wms_test_utils.WmsServicePassThru",
117 pass_thru="EXTRA_VALUES",
118 )
121class TestCommandStatus(unittest.TestCase):
122 """Test executing the status subcommand."""
124 def setUp(self):
125 self.runner = LogCliRunner()
127 def testStatusSuccess(self):
128 with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.status_driver") as mock_driver:
129 mock_driver.return_value = 0
130 result = self.runner.invoke(
131 bps.cli,
132 [
133 "status",
134 "--wms-service-class",
135 "wms_test_utils.WmsServiceDummy",
136 "--id",
137 "100",
138 ],
139 )
140 self.assertEqual(result.exit_code, 0)
141 mock_driver.assert_called_with(
142 wms_service="wms_test_utils.WmsServiceDummy",
143 run_id="100",
144 hist_days=0.0,
145 is_global=False,
146 )
148 def testStatusMissingIDCommandLine(self):
149 with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.status_driver") as mock_driver:
150 mock_driver.return_value = 0
151 result = self.runner.invoke(
152 bps.cli,
153 [
154 "status",
155 "--wms-service-class",
156 "wms_test_utils.WmsServiceDummy",
157 "--user",
158 "user_not_allowed",
159 ],
160 )
161 self.assertEqual(result.exit_code, 2)
163 def testStatusNonZeroStatus(self):
164 with unittest.mock.patch("lsst.ctrl.bps.cli.cmd.commands.status_driver") as mock_driver:
165 mock_driver.return_value = 15
166 result = self.runner.invoke(
167 bps.cli,
168 [
169 "status",
170 "--wms-service-class",
171 "wms_test_utils.WmsServiceDummy",
172 "--id",
173 "/dummy/path",
174 "--hist",
175 "6",
176 "--global",
177 ],
178 )
179 self.assertEqual(result.exit_code, 15)
180 mock_driver.assert_called_with(
181 wms_service="wms_test_utils.WmsServiceDummy",
182 run_id="/dummy/path",
183 hist_days=6.0,
184 is_global=True,
185 )
188if __name__ == "__main__":
189 unittest.main()