Coverage for tests / test_cli.py: 28%

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, text 

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") 

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

39 

40 

41class CliTestCase(unittest.TestCase): 

42 """Tests for CLI commands.""" 

43 

44 def setUp(self) -> None: 

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

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

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

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

49 

50 def tearDown(self) -> None: 

51 """Clean up temporary directory.""" 

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

53 

54 def test_invalid_command(self) -> None: 

55 """Test for invalid command.""" 

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

57 

58 def test_help(self) -> None: 

59 """Test for help command.""" 

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

61 

62 def test_create(self) -> None: 

63 """Test for create command.""" 

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

65 

66 def test_create_with_echo(self) -> None: 

67 """Test for create command.""" 

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

69 

70 def test_create_with_dry_run(self) -> None: 

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

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

73 

74 def test_create_with_ignore_constraints(self) -> None: 

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

76 run_cli( 

77 [ 

78 "create", 

79 "--schema-name=main", 

80 "--ignore-constraints", 

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

82 "--dry-run", 

83 TEST_YAML, 

84 ] 

85 ) 

86 

87 def test_validate(self) -> None: 

88 """Test validate command.""" 

89 run_cli(["validate", TEST_YAML]) 

90 

91 def test_validate_with_id_generation(self) -> None: 

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

93 enabled. This is the default behavior. 

94 """ 

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

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

97 

98 def test_validate_with_id_generation_error(self) -> None: 

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

100 enabled. 

101 """ 

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

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

104 

105 def test_validate_with_extra_checks(self) -> None: 

106 """Test schema validation flags.""" 

107 run_cli( 

108 [ 

109 "validate", 

110 "--check-description", 

111 "--check-tap-principal", 

112 "--check-tap-table-indexes", 

113 TEST_YAML, 

114 ] 

115 ) 

116 

117 def test_create_with_initialize_and_drop_error(self) -> None: 

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

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

120 

121 def test_load_tap_schema(self) -> None: 

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

123 # Create the TAP_SCHEMA database. 

124 tap_schema_path = tap_schema.TableManager.get_tap_schema_std_path() 

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

126 

127 # Load the TAP_SCHEMA data. 

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

129 

130 def test_load_tap_schema_with_dry_run_and_output_file(self) -> None: 

131 """Test load-tap-schema command with dry run and output file.""" 

132 output_sql = os.path.join(self.tmpdir, "tap_schema.sql") 

133 run_cli( 

134 [ 

135 "load-tap-schema", 

136 "--engine-url=mysql://", 

137 "--dry-run", 

138 "--tap-schema-index=1", 

139 "--tap-tables-postfix=11", 

140 "--force-unbounded-arraysize", 

141 f"--output-file={output_sql}", 

142 TEST_YAML, 

143 ] 

144 ) 

145 if not os.path.exists(output_sql): 

146 self.fail("Output SQL file was not created") 

147 if os.path.getsize(output_sql) == 0: 

148 self.fail("Output SQL file is empty") 

149 

150 def test_init_tap_schema(self) -> None: 

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

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

153 

154 def test_init_tap_schema_mock(self) -> None: 

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

156 an error, as this is not supported. 

157 """ 

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

159 

160 def test_init_tap_schema_with_extensions(self) -> None: 

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

162 run_cli( 

163 [ 

164 "init-tap-schema", 

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

166 "--extensions", 

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

168 ] 

169 ) 

170 

171 def test_init_tap_schema_with_custom_extensions(self) -> None: 

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

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

174 extensions_content = """ 

175 name: TAP_SCHEMA 

176 tables: 

177 - name: schemas 

178 columns: 

179 - name: field1 

180 datatype: char 

181 length: 64 

182 description: A custom field 

183 """ 

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

185 f.write(extensions_content) 

186 

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

188 

189 def test_diff(self) -> None: 

190 """Test for ``diff`` command.""" 

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

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

193 

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

195 

196 def test_diff_database(self) -> None: 

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

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

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

200 

201 engine = create_engine(self.sqlite_url) 

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

203 metadata_db.create_all(engine) 

204 engine.dispose() 

205 

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

207 

208 def test_diff_alembic(self) -> None: 

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

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

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

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

213 

214 def test_diff_error(self) -> None: 

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

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

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

218 

219 def test_diff_error_on_change(self) -> None: 

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

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

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

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

224 

225 def test_dump_yaml(self) -> None: 

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

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

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

229 

230 @classmethod 

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

232 """ 

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

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

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

236 """ 

237 if isinstance(obj, dict): 

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

239 if k == "@id": 

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

241 cls._check_strip_ids(v) 

242 elif isinstance(obj, list): 

243 for item in obj: 

244 cls._check_strip_ids(item) 

245 

246 def test_dump_yaml_with_strip_ids(self) -> None: 

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

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

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

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

251 try: 

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

253 data = yaml.safe_load(dumped_data) 

254 self._check_strip_ids(data) 

255 except ValueError: 

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

257 

258 def test_dump_json(self) -> None: 

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

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

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

262 

263 def test_dump_json_with_strip_ids(self) -> None: 

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

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

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

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

268 try: 

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

270 data = yaml.safe_load(dumped_data) 

271 self._check_strip_ids(data) 

272 except ValueError: 

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

274 

275 def test_dump_with_invalid_file_extension_error(self) -> None: 

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

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

278 

279 def test_create_and_drop_indexes(self) -> None: 

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

281 SQLite; no checking for the existence of the indexes is done on the 

282 database because other test cases cover that functionality 

283 sufficiently. 

284 """ 

285 # Create database without indexes 

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

287 

288 # Create the indexes using CLI 

289 run_cli( 

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

291 ) 

292 

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

294 run_cli( 

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

296 ) 

297 

298 # Drop the indexes using CLI 

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

300 

301 def test_generate_and_load_sql(self) -> None: 

302 """Test generating SQL and then executing it on a SQLite database.""" 

303 generated_sql = os.path.join(self.tmpdir, "generated.sql") 

304 

305 try: 

306 # Generate SQL DDL from schema using mock connection 

307 run_cli( 

308 [ 

309 "create", 

310 "--engine-url=sqlite://", 

311 f"--output-file={generated_sql}", 

312 f"{TEST_YAML}", 

313 ] 

314 ) 

315 

316 # Verify the SQL file was generated 

317 self.assertTrue(os.path.exists(generated_sql), "Generated SQL file should exist") 

318 

319 # Read the generated SQL 

320 with open(generated_sql) as f: 

321 sql = f.read() 

322 

323 # Verify SQL content is not empty 

324 self.assertGreater(len(sql.strip()), 0, "Generated SQL should not be empty") 

325 

326 # Execute the SQL against a real database 

327 engine = create_engine(self.sqlite_url) 

328 with engine.connect() as connection: 

329 with connection.begin(): 

330 # Split SQL into individual statements for execution since 

331 # SQLite can only execute one statement at a time 

332 statements = [stmt.strip() for stmt in sql.split(";") if stmt.strip()] 

333 for statement in statements: 

334 if statement: # Skip empty statements 

335 connection.execute(text(statement)) 

336 

337 # Verify that all expected tables were actually created 

338 with engine.connect() as connection: 

339 # Load the schema to get expected table names 

340 schema = Schema.from_uri(TEST_YAML, context={"id_generation": True}) 

341 expected_table_names = {table.name for table in schema.tables} 

342 

343 # Get all tables that were created in the database 

344 result = connection.execute(text("SELECT name FROM sqlite_master WHERE type='table'")) 

345 created_table_names = {row[0] for row in result.fetchall()} 

346 

347 # Verify all expected tables were created 

348 self.assertTrue( 

349 expected_table_names.issubset(created_table_names), 

350 f"Missing tables: {expected_table_names - created_table_names}. " 

351 f"Expected: {sorted(expected_table_names)}, " 

352 f"Created: {sorted(created_table_names)}", 

353 ) 

354 

355 engine.dispose() 

356 

357 except Exception as e: 

358 self.fail(f"Test failed with exception: {e}") 

359 

360 

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

362 unittest.main()