Coverage for tests/test_cli.py: 35%

1# This file is part of felis. 

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

21 

22import os 

23import shutil 

24import tempfile 

25import unittest 

26from typing import Any 

27 

28import yaml 

29from sqlalchemy import create_engine 

30 

31import felis.tap_schema as tap_schema 

32from felis.datamodel import Schema 

33from felis.metadata import MetaDataBuilder 

34from felis.tests.run_cli import run_cli 

35 

36TEST_DIR = os.path.abspath(os.path.dirname(__file__)) 

37TEST_YAML = os.path.join(TEST_DIR, "data", "test.yml") 

38 

39 

40class CliTestCase(unittest.TestCase): 

41 """Tests for CLI commands.""" 

42 

43 def setUp(self) -> None: 

44 """Set up a temporary directory for tests.""" 

45 self.tmpdir = tempfile.mkdtemp(dir=TEST_DIR) 

46 self.sqlite_url = f"sqlite:///{self.tmpdir}/db.sqlite3" 

47 print(f"Using temporary directory: {self.tmpdir}") 

48 

49 def tearDown(self) -> None: 

50 """Clean up temporary directory.""" 

51 shutil.rmtree(self.tmpdir, ignore_errors=True) 

52 

53 def test_invalid_command(self) -> None: 

54 """Test for invalid command.""" 

55 run_cli(["invalid"], expect_error=True) 

56 

57 def test_help(self) -> None: 

58 """Test for help command.""" 

59 run_cli(["--help"], print_output=True) 

60 

61 def test_create(self) -> None: 

62 """Test for create command.""" 

63 run_cli(["create", f"--engine-url={self.sqlite_url}", TEST_YAML]) 

64 

65 def test_create_with_dry_run(self) -> None: 

66 """Test for ``create --dry-run`` command.""" 

67 run_cli(["create", "--schema-name=main", f"--engine-url={self.sqlite_url}", "--dry-run", TEST_YAML]) 

68 

69 def test_create_with_ignore_constraints(self) -> None: 

70 """Test ``--ignore-constraints`` flag of ``create`` command.""" 

71 run_cli( 

72 [ 

73 "create", 

74 "--schema-name=main", 

75 "--ignore-constraints", 

76 f"--engine-url={self.sqlite_url}", 

77 "--dry-run", 

78 TEST_YAML, 

79 ] 

80 ) 

81 

82 def test_validate(self) -> None: 

83 """Test validate command.""" 

84 run_cli(["validate", TEST_YAML]) 

85 

86 def test_validate_with_id_generation(self) -> None: 

87 """Test that loading a schema with IDs works if ID generation is 

88 enabled. This is the default behavior. 

89 """ 

90 test_yaml = os.path.join(TEST_DIR, "data", "test_id_generation.yaml") 

91 run_cli(["--id-generation", "validate", test_yaml]) 

92 

93 def test_validate_with_id_generation_error(self) -> None: 

94 """Test that loading a schema without IDs fails if ID generation is not 

95 enabled. 

96 """ 

97 test_yaml = os.path.join(TEST_DIR, "data", "test_id_generation.yaml") 

98 run_cli(["--no-id-generation", "validate", test_yaml], expect_error=True) 

99 

100 def test_validate_with_extra_checks(self) -> None: 

101 """Test schema validation flags.""" 

102 run_cli( 

103 [ 

104 "validate", 

105 "--check-description", 

106 "--check-tap-principal", 

107 "--check-tap-table-indexes", 

108 TEST_YAML, 

109 ] 

110 ) 

111 

112 def test_create_with_initialize_and_drop_error(self) -> None: 

113 """Test that initialize and drop can't be used together.""" 

114 run_cli(["create", "--initialize", "--drop", TEST_YAML], expect_error=True) 

115 

116 def test_load_tap_schema(self) -> None: 

117 """Test load-tap-schema command.""" 

118 # Create the TAP_SCHEMA database. 

119 tap_schema_path = tap_schema.TableManager.get_tap_schema_std_path() 

120 run_cli(["--id-generation", "create", f"--engine-url={self.sqlite_url}", tap_schema_path]) 

121 

122 # Load the TAP_SCHEMA data. 

123 run_cli(["load-tap-schema", f"--engine-url={self.sqlite_url}", TEST_YAML]) 

124 

125 def test_init_tap_schema(self) -> None: 

126 """Test init-tap-schema command.""" 

127 run_cli(["init-tap-schema", f"--engine-url={self.sqlite_url}"]) 

128 

129 def test_init_tap_schema_mock(self) -> None: 

130 """Test init-tap-schema command with a mock URL, which should throw 

131 an error, as this is not supported. 

132 """ 

133 run_cli(["init-tap-schema", "sqlite://"], expect_error=True) 

134 

135 def test_diff(self) -> None: 

136 """Test for ``diff`` command.""" 

137 test_diff1 = os.path.join(TEST_DIR, "data", "test_diff1.yaml") 

138 test_diff2 = os.path.join(TEST_DIR, "data", "test_diff2.yaml") 

139 

140 run_cli(["diff", test_diff1, test_diff2]) 

141 

142 def test_diff_database(self) -> None: 

143 """Test for ``diff`` command with database.""" 

144 test_diff1 = os.path.join(TEST_DIR, "data", "test_diff1.yaml") 

145 test_diff2 = os.path.join(TEST_DIR, "data", "test_diff2.yaml") 

146 

147 engine = create_engine(self.sqlite_url) 

148 metadata_db = MetaDataBuilder(Schema.from_uri(test_diff1), apply_schema_to_metadata=False).build() 

149 metadata_db.create_all(engine) 

150 

151 run_cli(["diff", f"--engine-url={self.sqlite_url}", test_diff2]) 

152 

153 def test_diff_alembic(self) -> None: 

154 """Test for ``diff`` command with ``--alembic`` comparator option.""" 

155 test_diff1 = os.path.join(TEST_DIR, "data", "test_diff1.yaml") 

156 test_diff2 = os.path.join(TEST_DIR, "data", "test_diff2.yaml") 

157 run_cli(["diff", "--comparator", "alembic", test_diff1, test_diff2], print_output=True) 

158 

159 def test_diff_error(self) -> None: 

160 """Test for ``diff`` command with bad arguments.""" 

161 test_diff1 = os.path.join(TEST_DIR, "data", "test_diff1.yaml") 

162 run_cli(["diff", test_diff1], expect_error=True) 

163 

164 def test_diff_error_on_change(self) -> None: 

165 """Test for ``diff`` command with ``--error-on-change`` flag.""" 

166 test_diff1 = os.path.join(TEST_DIR, "data", "test_diff1.yaml") 

167 test_diff2 = os.path.join(TEST_DIR, "data", "test_diff2.yaml") 

168 run_cli(["diff", "--error-on-change", test_diff1, test_diff2], expect_error=True, print_output=True) 

169 

170 def test_dump_yaml(self) -> None: 

171 """Test for ``dump`` command with YAML output.""" 

172 with tempfile.NamedTemporaryFile(delete=True, suffix=".yaml") as temp_file: 

173 run_cli(["dump", TEST_YAML, temp_file.name], print_output=True) 

174 

175 @classmethod 

176 def _check_strip_ids(cls, obj: Any) -> None: 

177 """ 

178 Recursively check that a dict/list structure has no attributes with key 

179 '@id'. Raises a ValueError if any '@id' key is found. This is used to 

180 check the output of the `--strip-ids` option in the `dump` command. 

181 """ 

182 if isinstance(obj, dict): 

183 for k, v in obj.items(): 

184 if k == "@id": 

185 raise ValueError("Found forbidden key '@id'") 

186 cls._check_strip_ids(v) 

187 elif isinstance(obj, list): 

188 for item in obj: 

189 cls._check_strip_ids(item) 

190 

191 def test_dump_yaml_with_strip_ids(self) -> None: 

192 """Test for ``dump`` command with YAML output and stripped IDs.""" 

193 with tempfile.NamedTemporaryFile(delete=True, suffix=".yaml") as temp_file: 

194 run_cli(["dump", "--strip-ids", TEST_YAML, temp_file.name], print_output=True) 

195 dumped_data = temp_file.read().decode("utf-8") 

196 try: 

197 # Load the dumped YAML data to check for '@id' keys. 

198 data = yaml.safe_load(dumped_data) 

199 self._check_strip_ids(data) 

200 except ValueError: 

201 self.fail("Dumped YAML contains forbidden key '@id'") 

202 

203 def test_dump_json(self) -> None: 

204 """Test for ``dump`` command with JSON output.""" 

205 with tempfile.NamedTemporaryFile(delete=True, suffix=".json") as temp_file: 

206 run_cli(["dump", TEST_YAML, temp_file.name], print_output=True) 

207 

208 def test_dump_json_with_strip_ids(self) -> None: 

209 """Test for ``dump`` command with JSON output.""" 

210 with tempfile.NamedTemporaryFile(delete=True, suffix=".json") as temp_file: 

211 run_cli(["dump", "--strip-ids", TEST_YAML, temp_file.name], print_output=True) 

212 dumped_data = temp_file.read().decode("utf-8") 

213 try: 

214 # Load the dumped YAML data to check for '@id' keys. 

215 data = yaml.safe_load(dumped_data) 

216 self._check_strip_ids(data) 

217 except ValueError: 

218 self.fail("Dumped YAML contains forbidden key '@id'") 

219 

220 def test_dump_with_invalid_file_extension_error(self) -> None: 

221 """Test for ``dump`` command with JSON output.""" 

222 run_cli(["dump", TEST_YAML, "out.bad"], expect_error=True) 

223 

224 

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

226 unittest.main()