Coverage for tests/test_cli.py: 30%

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.db.utils import DatabaseContext 

34from felis.metadata import MetaDataBuilder 

35from felis.tests.run_cli import run_cli 

36 

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

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

39TEST_SALES_YAML = os.path.join(TEST_DIR, "data", "sales.yaml") 

40 

41 

42class CliTestCase(unittest.TestCase): 

43 """Tests for CLI commands.""" 

44 

45 def setUp(self) -> None: 

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

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

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

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

50 

51 def tearDown(self) -> None: 

52 """Clean up temporary directory.""" 

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

54 

55 def test_invalid_command(self) -> None: 

56 """Test for invalid command.""" 

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

58 

59 def test_help(self) -> None: 

60 """Test for help command.""" 

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

62 

63 def test_create(self) -> None: 

64 """Test for create command.""" 

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

66 

67 def test_create_with_dry_run(self) -> None: 

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

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

70 

71 def test_create_with_ignore_constraints(self) -> None: 

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

73 run_cli( 

74 [ 

75 "create", 

76 "--schema-name=main", 

77 "--ignore-constraints", 

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

79 "--dry-run", 

80 TEST_YAML, 

81 ] 

82 ) 

83 

84 def test_validate(self) -> None: 

85 """Test validate command.""" 

86 run_cli(["validate", TEST_YAML]) 

87 

88 def test_validate_with_id_generation(self) -> None: 

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

90 enabled. This is the default behavior. 

91 """ 

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

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

94 

95 def test_validate_with_id_generation_error(self) -> None: 

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

97 enabled. 

98 """ 

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

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

101 

102 def test_validate_with_extra_checks(self) -> None: 

103 """Test schema validation flags.""" 

104 run_cli( 

105 [ 

106 "validate", 

107 "--check-description", 

108 "--check-tap-principal", 

109 "--check-tap-table-indexes", 

110 TEST_YAML, 

111 ] 

112 ) 

113 

114 def test_create_with_initialize_and_drop_error(self) -> None: 

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

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

117 

118 def test_load_tap_schema(self) -> None: 

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

120 # Create the TAP_SCHEMA database. 

121 tap_schema_path = tap_schema.TableManager.get_tap_schema_std_path() 

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

123 

124 # Load the TAP_SCHEMA data. 

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

126 

127 def test_init_tap_schema(self) -> None: 

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

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

130 

131 def test_init_tap_schema_mock(self) -> None: 

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

133 an error, as this is not supported. 

134 """ 

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

136 

137 def test_init_tap_schema_with_extensions(self) -> None: 

138 """Test init-tap-schema command with default extensions.""" 

139 run_cli( 

140 [ 

141 "init-tap-schema", 

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

143 "--extensions", 

144 "resource://felis/config/tap_schema/tap_schema_extensions.yaml", 

145 ] 

146 ) 

147 

148 def test_init_tap_schema_with_custom_extensions(self) -> None: 

149 """Test init-tap-schema command with custom extensions file.""" 

150 extensions_file = os.path.join(self.tmpdir, "custom_extensions.yaml") 

151 extensions_content = """ 

152 name: TAP_SCHEMA 

153 tables: 

154 - name: schemas 

155 columns: 

156 - name: field1 

157 datatype: char 

158 length: 64 

159 description: A custom field 

160 """ 

161 with open(extensions_file, "w") as f: 

162 f.write(extensions_content) 

163 

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

165 

166 def test_diff(self) -> None: 

167 """Test for ``diff`` command.""" 

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

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

170 

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

172 

173 def test_diff_database(self) -> None: 

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

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

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

177 

178 engine = create_engine(self.sqlite_url) 

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

180 metadata_db.create_all(engine) 

181 

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

183 

184 def test_diff_alembic(self) -> None: 

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

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

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

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

189 

190 def test_diff_error(self) -> None: 

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

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

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

194 

195 def test_diff_error_on_change(self) -> None: 

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

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

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

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

200 

201 def test_dump_yaml(self) -> None: 

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

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

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

205 

206 @classmethod 

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

208 """ 

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

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

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

212 """ 

213 if isinstance(obj, dict): 

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

215 if k == "@id": 

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

217 cls._check_strip_ids(v) 

218 elif isinstance(obj, list): 

219 for item in obj: 

220 cls._check_strip_ids(item) 

221 

222 def test_dump_yaml_with_strip_ids(self) -> None: 

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

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

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

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

227 try: 

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

229 data = yaml.safe_load(dumped_data) 

230 self._check_strip_ids(data) 

231 except ValueError: 

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

233 

234 def test_dump_json(self) -> None: 

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

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

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

238 

239 def test_dump_json_with_strip_ids(self) -> None: 

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

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

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

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

244 try: 

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

246 data = yaml.safe_load(dumped_data) 

247 self._check_strip_ids(data) 

248 except ValueError: 

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

250 

251 def test_dump_with_invalid_file_extension_error(self) -> None: 

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

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

254 

255 def test_create_and_drop_indexes(self) -> None: 

256 """Test creating and dropping indexes using CLI commands with 

257 SQLite. 

258 """ 

259 # Load the schema to get expected indexes 

260 schema = Schema.from_uri(TEST_SALES_YAML) 

261 md_with_indexes = MetaDataBuilder(schema, skip_indexes=False).build() 

262 

263 engine = create_engine(self.sqlite_url) 

264 

265 def check_indexes_exist(should_exist: bool, message: str) -> None: 

266 """Check if indexes exist or don't exist in the database.""" 

267 with engine.connect() as conn: 

268 for table in md_with_indexes.tables.values(): 

269 for index in table.indexes: 

270 if index.name is not None: 

271 exists = DatabaseContext._index_exists(conn, table, index) 

272 if should_exist: 

273 self.assertTrue( 

274 exists, 

275 f"Index '{index.name}' {message}", 

276 ) 

277 else: 

278 self.assertFalse( 

279 exists, 

280 f"Index '{index.name}' {message}", 

281 ) 

282 

283 # Create database without indexes 

284 run_cli(["create", "--skip-indexes", f"--engine-url={self.sqlite_url}", TEST_SALES_YAML]) 

285 

286 # Check that indexes don't exist yet 

287 check_indexes_exist(False, "should not exist after create with --skip-indexes") 

288 

289 # Create the indexes using CLI 

290 run_cli( 

291 ["create-indexes", f"--engine-url={self.sqlite_url}", TEST_SALES_YAML, "--schema-name", "main"] 

292 ) 

293 

294 # Check that indexes now exist 

295 check_indexes_exist(True, "should exist after create-indexes") 

296 

297 # Create the indexes again; should not cause an error 

298 run_cli( 

299 ["create-indexes", f"--engine-url={self.sqlite_url}", TEST_SALES_YAML, "--schema-name", "main"] 

300 ) 

301 

302 # Drop the indexes using CLI 

303 run_cli(["drop-indexes", f"--engine-url={self.sqlite_url}", TEST_SALES_YAML, "--schema-name", "main"]) 

304 

305 # Check that indexes were dropped 

306 check_indexes_exist(False, "should not exist after drop-indexes") 

307 

308 

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

310 unittest.main()