Coverage for tests / test_cliPluginLoader.py: 36%

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/>. 

27 

28"""Unit tests for the daf_butler CLI plugin loader.""" 

29 

30import os 

31import unittest 

32from collections import defaultdict 

33from contextlib import contextmanager 

34from unittest.mock import patch 

35 

36import click 

37import yaml 

38 

39from lsst.daf.butler.cli import butler, cmd 

40from lsst.daf.butler.cli.butler import UncachedButlerCLI 

41from lsst.daf.butler.cli.utils import LogCliRunner, command_test_env 

42 

43 

44@click.command() 

45def command_test(): 

46 """Run command test.""" 

47 click.echo(message="test command") 

48 

49 

50@contextmanager 

51def duplicate_command_test_env(runner): 

52 """Context manager that creates (and then cleans up) an environment that 

53 declares a plugin command named 'create', which will conflict with the 

54 daf_butler 'create' command. 

55 

56 Parameters 

57 ---------- 

58 runner : click.testing.CliRunner 

59 The test runner to use to create the isolated filesystem. 

60 """ 

61 with runner.isolated_filesystem(): 

62 with open("resources.yaml", "w") as f: 

63 f.write(yaml.dump({"cmd": {"import": "test_cliPluginLoader", "commands": ["create"]}})) 

64 with patch.dict("os.environ", {"DAF_BUTLER_PLUGINS": os.path.realpath(f.name)}): 

65 yield 

66 

67 

68@click.command(cls=UncachedButlerCLI) 

69def uncached_cli(): 

70 """ButlerCLI that does not cache the commands.""" 

71 pass 

72 

73 

74class FailedLoadTest(unittest.TestCase): 

75 """Test failed plugin loading.""" 

76 

77 def setUp(self): 

78 self.runner = LogCliRunner() 

79 

80 def test_unimportablePlugin(self): 

81 with command_test_env(self.runner, "test_cliPluginLoader", "non-existant-command-function"): 

82 with self.assertLogs() as cm: 

83 result = self.runner.invoke(uncached_cli, "--help") 

84 self.assertEqual(result.exit_code, 0, f"output: {result.output!r} exception: {result.exception}") 

85 expectedErrMsg = ( 

86 "Could not import plugin from test_cliPluginLoader.non_existant_command_function, skipping." 

87 ) 

88 self.assertIn(expectedErrMsg, " ".join(cm.output)) 

89 

90 def test_unimportableLocalPackage(self): 

91 class FailCLI(butler.LoaderCLI): 

92 localCmdPkg = "lsst.daf.butler.cli.cmds" # should not be an importable location 

93 

94 @click.command(cls=FailCLI) 

95 def cli(): 

96 pass 

97 

98 with self.assertLogs() as cm: 

99 result = self.runner.invoke(cli, "--help") 

100 self.assertEqual(result.exit_code, 0, f"output: {result.output} exception: {result.exception}") 

101 expectedErrMsg = f"Could not import plugin from {FailCLI.localCmdPkg}, skipping." 

102 self.assertIn(expectedErrMsg, " ".join(cm.output)) 

103 

104 

105class PluginLoaderTest(unittest.TestCase): 

106 """Test the command-line plugin loader.""" 

107 

108 def setUp(self): 

109 self.runner = LogCliRunner() 

110 

111 def test_loadAndExecutePluginCommand(self): 

112 """Test that a plugin command can be loaded and executed.""" 

113 with command_test_env(self.runner, "test_cliPluginLoader", "command-test"): 

114 result = self.runner.invoke(uncached_cli, "command-test") 

115 self.assertEqual(result.exit_code, 0, f"output: {result.output} exception: {result.exception}") 

116 self.assertEqual(result.stdout, "test command\n") 

117 

118 def test_loadAndExecuteLocalCommand(self): 

119 """Test that a command in daf_butler can be loaded and executed.""" 

120 with self.runner.isolated_filesystem(): 

121 result = self.runner.invoke(butler.cli, ["create", "test_repo"]) 

122 self.assertEqual(result.exit_code, 0, f"output: {result.output} exception: {result.exception}") 

123 self.assertTrue(os.path.exists("test_repo")) 

124 

125 def test_loadTopHelp(self): 

126 """Test that an expected command is produced by 'butler --help'""" 

127 with command_test_env(self.runner, "test_cliPluginLoader", "command-test"): 

128 result = self.runner.invoke(uncached_cli, "--help") 

129 self.assertEqual(result.exit_code, 0, f"output: {result.output} exception: {result.exception}") 

130 self.assertIn("command-test", result.stdout) 

131 

132 def test_getLocalCommands(self): 

133 """Test getting the daf_butler CLI commands.""" 

134 localCommands = butler.ButlerCLI().getLocalCommands() 

135 # the number of local commands should equal the number of functions 

136 # in cmd.__all__ 

137 self.assertEqual(len(localCommands), len(cmd.__all__)) 

138 

139 def test_mergeCommandLists(self): 

140 """Verify dicts of command to list-of-source-package get merged 

141 properly. 

142 """ 

143 first = defaultdict(list, {"a": [1]}) 

144 second = defaultdict(list, {"b": [2]}) 

145 self.assertEqual(butler.LoaderCLI._mergeCommandLists(first, second), {"a": [1], "b": [2]}) 

146 first = defaultdict(list, {"a": [1]}) 

147 second = defaultdict(list, {"a": [2]}) 

148 self.assertEqual(butler.LoaderCLI._mergeCommandLists(first, second), {"a": [1, 2]}) 

149 

150 def test_listCommands_duplicate(self): 

151 """Test executing a command in a situation where duplicate commands are 

152 present and verify it fails to run. 

153 """ 

154 self.maxDiff = None 

155 with duplicate_command_test_env(self.runner): 

156 result = self.runner.invoke(uncached_cli, ["create", "test_repo"]) 

157 self.assertEqual(result.exit_code, 1, f"output: {result.output} exception: {result.exception}") 

158 self.assertEqual( 

159 result.output, 

160 "Error: Command 'create' " 

161 "exists in packages lsst.daf.butler.cli.cmd, test_cliPluginLoader. " 

162 "Duplicate commands are not supported, aborting.\n", 

163 ) 

164 

165 

166if __name__ == "__main__": 

167 unittest.main()