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_diff(self) -> None: 

138 """Test for ``diff`` command.""" 

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

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

141 

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

143 

144 def test_diff_database(self) -> None: 

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

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

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

148 

149 engine = create_engine(self.sqlite_url) 

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

151 metadata_db.create_all(engine) 

152 

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

154 

155 def test_diff_alembic(self) -> None: 

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

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

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

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

160 

161 def test_diff_error(self) -> None: 

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

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

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

165 

166 def test_diff_error_on_change(self) -> None: 

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

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 run_cli(["diff", "--error-on-change", test_diff1, test_diff2], expect_error=True, print_output=True) 

171 

172 def test_dump_yaml(self) -> None: 

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

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

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

176 

177 @classmethod 

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

179 """ 

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

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

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

183 """ 

184 if isinstance(obj, dict): 

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

186 if k == "@id": 

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

188 cls._check_strip_ids(v) 

189 elif isinstance(obj, list): 

190 for item in obj: 

191 cls._check_strip_ids(item) 

192 

193 def test_dump_yaml_with_strip_ids(self) -> None: 

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

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

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

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

198 try: 

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

200 data = yaml.safe_load(dumped_data) 

201 self._check_strip_ids(data) 

202 except ValueError: 

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

204 

205 def test_dump_json(self) -> None: 

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

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

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

209 

210 def test_dump_json_with_strip_ids(self) -> None: 

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

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

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

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

215 try: 

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

217 data = yaml.safe_load(dumped_data) 

218 self._check_strip_ids(data) 

219 except ValueError: 

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

221 

222 def test_dump_with_invalid_file_extension_error(self) -> None: 

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

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

225 

226 def test_create_and_drop_indexes(self) -> None: 

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

228 SQLite. 

229 """ 

230 # Load the schema to get expected indexes 

231 schema = Schema.from_uri(TEST_SALES_YAML) 

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

233 

234 engine = create_engine(self.sqlite_url) 

235 

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

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

238 with engine.connect() as conn: 

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

240 for index in table.indexes: 

241 if index.name is not None: 

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

243 if should_exist: 

244 self.assertTrue( 

245 exists, 

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

247 ) 

248 else: 

249 self.assertFalse( 

250 exists, 

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

252 ) 

253 

254 # Create database without indexes 

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

256 

257 # Check that indexes don't exist yet 

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

259 

260 # Create the indexes using CLI 

261 run_cli( 

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

263 ) 

264 

265 # Check that indexes now exist 

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

267 

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

269 run_cli( 

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

271 ) 

272 

273 # Drop the indexes using CLI 

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

275 

276 # Check that indexes were dropped 

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

278 

279 

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

281 unittest.main()