Coverage for tests/test_metadata.py: 10%

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 unittest 

24 

25import yaml 

26from sqlalchemy import ( 

27 CheckConstraint, 

28 Constraint, 

29 ForeignKeyConstraint, 

30 Index, 

31 MetaData, 

32 PrimaryKeyConstraint, 

33 UniqueConstraint, 

34 create_engine, 

35) 

36 

37from felis import datamodel as dm 

38from felis.datamodel import Schema 

39from felis.db.utils import DatabaseContext 

40from felis.metadata import MetaDataBuilder, get_datatype_with_variants 

41 

42TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

43TEST_YAML = os.path.join(TESTDIR, "data", "sales.yaml") 

44 

45 

46class MetaDataTestCase(unittest.TestCase): 

47 """Test creation of SQLAlchemy metadata from a Felis schema.""" 

48 

49 def setUp(self) -> None: 

50 """Create an in-memory SQLite database and load the test data.""" 

51 self.engine = create_engine("sqlite://") 

52 with open(TEST_YAML) as data: 

53 self.yaml_data = yaml.safe_load(data) 

54 

55 def test_create_all(self) -> None: 

56 """Create all tables in the schema using the metadata object and a 

57 SQLite connection. 

58 

59 Check that the reflected metadata matches that created by the builder. 

60 """ 

61 

62 def _sorted_indexes(indexes: set[Index]) -> list[Index]: 

63 """Return a sorted list of indexes.""" 

64 return sorted(indexes, key=lambda i: i.name) 

65 

66 def _sorted_constraints(constraints: set[Constraint]) -> list[Constraint]: 

67 """Return a sorted list of constraints with the 

68 `PrimaryKeyConstraint` objects filtered out. 

69 """ 

70 return sorted( 

71 [c for c in constraints if not isinstance(c, PrimaryKeyConstraint)], key=lambda c: c.name 

72 ) 

73 

74 schema = Schema.model_validate(self.yaml_data) 

75 schema.name = "main" 

76 builder = MetaDataBuilder(schema) 

77 md = builder.build() 

78 

79 ctx = DatabaseContext(md, self.engine) 

80 

81 ctx.create_all() 

82 

83 md_db = MetaData() 

84 md_db.reflect(self.engine.connect(), schema=schema.name) 

85 

86 self.assertEqual(md_db.tables.keys(), md.tables.keys()) 

87 

88 for md_table_name in md.tables.keys(): 

89 md_table = md.tables[md_table_name] 

90 md_db_table = md_db.tables[md_table_name] 

91 self.assertEqual(md_table.columns.keys(), md_db_table.columns.keys()) 

92 for md_column_name in md_table.columns.keys(): 

93 md_column = md_table.columns[md_column_name] 

94 md_db_column = md_db_table.columns[md_column_name] 

95 self.assertEqual(type(md_column.type), type(md_db_column.type)) 

96 self.assertEqual(md_column.nullable, md_db_column.nullable) 

97 self.assertEqual(md_column.primary_key, md_db_column.primary_key) 

98 self.assertTrue( 

99 (md_table.constraints and md_db_table.constraints) 

100 or (not md_table.constraints and not md_table.constraints), 

101 "Constraints not created correctly", 

102 ) 

103 if md_table.constraints: 

104 self.assertEqual(len(md_table.constraints), len(md_db_table.constraints)) 

105 md_constraints = _sorted_constraints(md_table.constraints) 

106 md_db_constraints = _sorted_constraints(md_db_table.constraints) 

107 for md_constraint, md_db_constraint in zip(md_constraints, md_db_constraints): 

108 self.assertEqual(md_constraint.name, md_db_constraint.name) 

109 self.assertEqual(md_constraint.deferrable, md_db_constraint.deferrable) 

110 self.assertEqual(md_constraint.initially, md_db_constraint.initially) 

111 self.assertEqual( 

112 type(md_constraint), type(md_db_constraint), "Constraint types do not match" 

113 ) 

114 if isinstance(md_constraint, ForeignKeyConstraint) and isinstance( 

115 md_db_constraint, ForeignKeyConstraint 

116 ): 

117 md_fk: ForeignKeyConstraint = md_constraint 

118 md_db_fk: ForeignKeyConstraint = md_db_constraint 

119 self.assertEqual(md_fk.referred_table.name, md_db_fk.referred_table.name) 

120 self.assertEqual(md_fk.column_keys, md_db_fk.column_keys) 

121 self.assertEqual(md_fk.ondelete, md_db_fk.ondelete) 

122 self.assertEqual(md_fk.onupdate, md_db_fk.onupdate) 

123 elif isinstance(md_constraint, UniqueConstraint) and isinstance( 

124 md_db_constraint, UniqueConstraint 

125 ): 

126 md_uniq: UniqueConstraint = md_constraint 

127 md_db_uniq: UniqueConstraint = md_db_constraint 

128 self.assertEqual(md_uniq.columns.keys(), md_db_uniq.columns.keys()) 

129 elif isinstance(md_constraint, CheckConstraint) and isinstance( 

130 md_db_constraint, CheckConstraint 

131 ): 

132 md_check: CheckConstraint = md_constraint 

133 md_db_check: CheckConstraint = md_db_constraint 

134 self.assertEqual(str(md_check.sqltext), str(md_db_check.sqltext)) 

135 self.assertTrue( 

136 (md_table.indexes and md_db_table.indexes) or (not md_table.indexes and not md_table.indexes), 

137 "Indexes not created correctly", 

138 ) 

139 if md_table.indexes: 

140 md_indexes = _sorted_indexes(md_table.indexes) 

141 md_db_indexes = _sorted_indexes(md_db_table.indexes) 

142 self.assertEqual(len(md_indexes), len(md_db_indexes)) 

143 for md_index, md_db_index in zip(md_table.indexes, md_db_table.indexes): 

144 self.assertEqual(md_index.name, md_db_index.name) 

145 self.assertEqual(md_index.columns.keys(), md_db_index.columns.keys()) 

146 

147 def test_builder(self) -> None: 

148 """Test that the information in the metadata object created by the 

149 builder matches the data in the Felis schema used to create it. 

150 """ 

151 sch = Schema.model_validate(self.yaml_data) 

152 bld = MetaDataBuilder(sch, apply_schema_to_metadata=False) 

153 md = bld.build() 

154 

155 self.assertEqual(len(sch.tables), len(md.tables)) 

156 self.assertEqual([table.name for table in sch.tables], list(md.tables.keys())) 

157 for table in sch.tables: 

158 md_table = md.tables[table.name] 

159 self.assertEqual(table.name, md_table.name) 

160 self.assertEqual(len(table.columns), len(md_table.columns)) 

161 for column in table.columns: 

162 md_table_column = md_table.columns[column.name] 

163 datatype = get_datatype_with_variants(column) 

164 self.assertEqual(type(datatype), type(md_table_column.type)) 

165 if column.nullable is not None: 

166 self.assertEqual(column.nullable, md_table_column.nullable) 

167 for constraint in table.constraints: 

168 md_constraint = [mdc for mdc in md_table.constraints if mdc.name == constraint.name][0] 

169 if isinstance(constraint, dm.ForeignKeyConstraint): 

170 self.assertTrue(isinstance(md_constraint, ForeignKeyConstraint)) 

171 self.assertTrue( 

172 sorted([sch[column_id].name for column_id in constraint.columns]), 

173 sorted(md_constraint.columns.keys()), 

174 ) 

175 elif isinstance(constraint, dm.UniqueConstraint): 

176 self.assertEqual( 

177 sorted([sch[column_id].name for column_id in constraint.columns]), 

178 sorted(md_constraint.columns.keys()), 

179 ) 

180 elif isinstance(constraint, dm.CheckConstraint): 

181 self.assertEqual(constraint.expression, str(md_constraint.sqltext)) 

182 for index in table.indexes: 

183 md_index = [mdi for mdi in md_table.indexes if mdi.name == index.name][0] 

184 self.assertEqual( 

185 sorted([sch[column_id].name for column_id in index.columns]), 

186 sorted(md_index.columns.keys()), 

187 ) 

188 if table.primary_key: 

189 if isinstance(table.primary_key, str): 

190 primary_keys = [sch[table.primary_key].name] 

191 else: 

192 primary_keys = [sch[pk].name for pk in table.primary_key] 

193 for primary_key in primary_keys: 

194 self.assertTrue(md_table.columns[primary_key].primary_key) 

195 

196 def test_timestamp(self) -> None: 

197 """Test that the `timestamp` datatype is created correctly.""" 

198 for precision in [None, 6]: 

199 col = dm.Column( 

200 **{ 

201 "name": "timestamp_test", 

202 "id": "#timestamp_test", 

203 "datatype": "timestamp", 

204 "precision": precision, 

205 } 

206 ) 

207 datatype = get_datatype_with_variants(col) 

208 variant_dict = datatype._variant_mapping 

209 self.assertTrue("mysql" in variant_dict) 

210 self.assertTrue("postgresql" in variant_dict) 

211 pg_timestamp = variant_dict["postgresql"] 

212 self.assertEqual(pg_timestamp.timezone, False) 

213 self.assertEqual(pg_timestamp.precision, precision) 

214 mysql_timestamp = variant_dict["mysql"] 

215 self.assertEqual(mysql_timestamp.timezone, False) 

216 self.assertEqual(mysql_timestamp.fsp, precision) 

217 

218 def test_ignore_constraints(self) -> None: 

219 """Test that constraints are not created when the 

220 ``ignore_constraints`` flag is set on the metadata builder. 

221 """ 

222 schema = Schema.model_validate(self.yaml_data) 

223 schema.name = "main" 

224 builder = MetaDataBuilder(schema, ignore_constraints=True) 

225 md = builder.build() 

226 for table in md.tables.values(): 

227 non_primary_key_constraints = [ 

228 c for c in table.constraints if not isinstance(c, PrimaryKeyConstraint) 

229 ] 

230 self.assertEqual( 

231 len(non_primary_key_constraints), 

232 0, 

233 msg=f"Table {table.name} has non-primary key constraints defined", 

234 ) 

235 

236 def test_table_name_postfix(self) -> None: 

237 """Test that table name postfixes are correctly applied.""" 

238 schema = Schema.model_validate(self.yaml_data) 

239 schema.name = "main" 

240 builder = MetaDataBuilder(schema, table_name_postfix="_test") 

241 md = builder.build() 

242 for table in md.tables.values(): 

243 self.assertTrue(table.name.endswith("_test")) 

244 

245 def test_fk_actions(self) -> None: 

246 """Test that foreign key constraints with on delete and on update 

247 actions are created correctly. 

248 """ 

249 schema = Schema.model_validate(self.yaml_data) 

250 schema.name = "main" 

251 builder = MetaDataBuilder(schema) 

252 md = builder.build() 

253 

254 for table in md.tables.values(): 

255 for constraint in table.constraints: 

256 if isinstance(constraint, ForeignKeyConstraint): 

257 self.assertIn(constraint.ondelete, ["SET NULL"]) 

258 self.assertIn(constraint.onupdate, ["CASCADE"]) 

259 

260 

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

262 unittest.main()