Coverage for tests/test_apdbSqlSchema.py: 100%

1# This file is part of dax_apdb. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://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 <http://www.gnu.org/licenses/>. 

21 

22"""Unit test for ApdbSqlSchema class.""" 

23 

24import os 

25import unittest 

26from typing import Any 

27 

28import sqlalchemy 

29from sqlalchemy import create_engine 

30 

31import lsst.utils.tests 

32from lsst.dax.apdb.apdbSchema import ApdbSchema, ApdbTables 

33from lsst.dax.apdb.sql.apdbSqlSchema import ApdbSqlSchema, ExtraTables 

34from lsst.dax.apdb.tests import update_schema_yaml 

35 

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

37TEST_SCHEMA = os.path.join(_HERE, "config/schema-apdb.yaml") 

38TEST_SCHEMA_COMBINED = os.path.join(_HERE, "config/schema-apdb+sso.yaml") 

39TEST_SCHEMA_SSO = os.path.join(_HERE, "config/schema-sso.yaml") 

40 

41 

42class ApdbSchemaTestCase(unittest.TestCase): 

43 """Test case for ApdbSqlSchema class. 

44 

45 Schema is defined in YAML files, some checks here depend on that 

46 configuration and will need to be updated when configuration changes. 

47 """ 

48 

49 # number of columns as defined in tests/config/schema.yaml 

50 table_column_count = { 

51 ApdbTables.DiaObject: 8, 

52 ApdbTables.DiaObjectLast: 6, 

53 ApdbTables.DiaSource: 13, 

54 ApdbTables.DiaForcedSource: 9, 

55 ApdbTables.DiaObject_To_Object_Match: 3, 

56 ApdbTables.SSObject: 3, 

57 ApdbTables.SSSource: 4, 

58 ApdbTables.metadata: 2, 

59 } 

60 

61 def _assertTable(self, table: sqlalchemy.schema.Table, name: str, ncol: int) -> None: 

62 """Validate tables schema. 

63 

64 Parameters 

65 ---------- 

66 table : `sqlalchemy.Table` 

67 name : `str` 

68 Expected table name 

69 ncol : `int` 

70 Expected number of columns 

71 """ 

72 self.assertIsNotNone(table) 

73 self.assertEqual(table.name, name) 

74 self.assertEqual(len(table.columns), ncol) 

75 

76 def test_makeSchema_default(self) -> None: 

77 """Test for creating schema.""" 

78 engine = create_engine("sqlite://") 

79 self.addCleanup(engine.dispose) 

80 

81 # create standard (baseline) schema 

82 schema = ApdbSqlSchema( 

83 table_schema=ApdbSchema(TEST_SCHEMA, TEST_SCHEMA_SSO), 

84 engine=engine, 

85 dia_object_index="baseline", 

86 htm_index_column="pixelId", 

87 ) 

88 schema.makeSchema() 

89 

90 for apdb_table in ApdbTables: 

91 if apdb_table in (ApdbTables.DiaObjectLast, ApdbTables.SSObject, ApdbTables.SSSource): 

92 with self.assertRaisesRegex(ValueError, ".*does not exist in the schema"): 

93 table = schema.get_table(apdb_table) 

94 continue 

95 

96 table = schema.get_table(apdb_table) 

97 

98 column_count = self.table_column_count[apdb_table] 

99 if apdb_table in (ApdbTables.DiaObject, ApdbTables.DiaSource): 

100 # DiaObject and DiaSource tables add pixelId column. 

101 column_count += 1 

102 

103 self._assertTable(table, apdb_table.table_name(), column_count) 

104 self.assertEqual(len(schema.get_apdb_columns(apdb_table)), self.table_column_count[apdb_table]) 

105 if apdb_table is ApdbTables.DiaObject: 

106 self.assertEqual(len(table.primary_key), 2) 

107 

108 for table_enum in ExtraTables: 

109 with self.assertRaisesRegex(ValueError, ".*does not exist in the schema"): 

110 schema.get_table(table_enum) 

111 

112 def test_makeSchema_prefix(self) -> None: 

113 """Create schema using prefix.""" 

114 engine = create_engine("sqlite://") 

115 self.addCleanup(engine.dispose) 

116 schema = ApdbSqlSchema( 

117 table_schema=ApdbSchema(TEST_SCHEMA_COMBINED, ""), 

118 engine=engine, 

119 dia_object_index="baseline", 

120 htm_index_column="pixelId", 

121 prefix="Pfx", 

122 ) 

123 # Drop existing tables (but we don't check it here) 

124 schema.makeSchema(drop=True) 

125 self._assertTable( 

126 schema.get_table(ApdbTables.DiaObject), 

127 "PfxDiaObject", 

128 self.table_column_count[ApdbTables.DiaObject] + 1, 

129 ) 

130 with self.assertRaisesRegex(ValueError, ".*does not exist in the schema"): 

131 schema.get_table(ApdbTables.DiaObjectLast) 

132 self._assertTable( 

133 schema.get_table(ApdbTables.DiaSource), 

134 "PfxDiaSource", 

135 self.table_column_count[ApdbTables.DiaSource] + 1, 

136 ) 

137 self._assertTable( 

138 schema.get_table(ApdbTables.DiaForcedSource), 

139 "PfxDiaForcedSource", 

140 self.table_column_count[ApdbTables.DiaForcedSource], 

141 ) 

142 

143 def test_makeSchema_other_index(self) -> None: 

144 """Use different indexing for DiaObject, this changes number of PK 

145 columns. 

146 """ 

147 engine = create_engine("sqlite://") 

148 self.addCleanup(engine.dispose) 

149 schema = ApdbSqlSchema( 

150 table_schema=ApdbSchema(TEST_SCHEMA, TEST_SCHEMA_SSO), 

151 engine=engine, 

152 dia_object_index="pix_id_iov", 

153 htm_index_column="pixelId", 

154 ) 

155 schema.makeSchema(drop=True) 

156 table = schema.get_table(ApdbTables.DiaObject) 

157 self._assertTable(table, "DiaObject", self.table_column_count[ApdbTables.DiaObject] + 1) 

158 self.assertEqual(len(table.primary_key), 3) 

159 with self.assertRaisesRegex(ValueError, ".*does not exist in the schema"): 

160 schema.get_table(ApdbTables.DiaObjectLast) 

161 self._assertTable( 

162 schema.get_table(ApdbTables.DiaSource), 

163 "DiaSource", 

164 self.table_column_count[ApdbTables.DiaSource] + 1, 

165 ) 

166 self._assertTable( 

167 schema.get_table(ApdbTables.DiaForcedSource), 

168 "DiaForcedSource", 

169 self.table_column_count[ApdbTables.DiaForcedSource], 

170 ) 

171 

172 def test_makeSchema_diaobjectlast(self) -> None: 

173 """Use DiaObjectLast table for DiaObject.""" 

174 engine = create_engine("sqlite://") 

175 self.addCleanup(engine.dispose) 

176 schema = ApdbSqlSchema( 

177 table_schema=ApdbSchema(TEST_SCHEMA, TEST_SCHEMA_SSO), 

178 engine=engine, 

179 dia_object_index="last_object_table", 

180 htm_index_column="pixelId", 

181 ) 

182 schema.makeSchema(drop=True) 

183 table = schema.get_table(ApdbTables.DiaObject) 

184 self._assertTable(table, "DiaObject", self.table_column_count[ApdbTables.DiaObject] + 1) 

185 self.assertEqual(len(table.primary_key), 2) 

186 table = schema.get_table(ApdbTables.DiaObjectLast) 

187 self._assertTable(table, "DiaObjectLast", self.table_column_count[ApdbTables.DiaObjectLast] + 1) 

188 self.assertEqual(len(table.primary_key), 2) 

189 self._assertTable( 

190 schema.get_table(ApdbTables.DiaSource), 

191 "DiaSource", 

192 self.table_column_count[ApdbTables.DiaSource] + 1, 

193 ) 

194 self._assertTable( 

195 schema.get_table(ApdbTables.DiaForcedSource), 

196 "DiaForcedSource", 

197 self.table_column_count[ApdbTables.DiaForcedSource], 

198 ) 

199 

200 def test_makeSchema_replica(self) -> None: 

201 """Add replica tables.""" 

202 engine = create_engine("sqlite://") 

203 self.addCleanup(engine.dispose) 

204 schema = ApdbSqlSchema( 

205 table_schema=ApdbSchema(TEST_SCHEMA, TEST_SCHEMA_SSO), 

206 engine=engine, 

207 dia_object_index="last_object_table", 

208 htm_index_column="pixelId", 

209 enable_replica=True, 

210 ) 

211 schema.makeSchema(drop=True) 

212 self._assertTable(schema.get_table(ExtraTables.ApdbReplicaChunks), "ApdbReplicaChunks", 3) 

213 self.assertEqual(len(schema.get_apdb_columns(ExtraTables.ApdbReplicaChunks)), 3) 

214 self._assertTable(schema.get_table(ExtraTables.DiaObjectChunks), "DiaObjectChunks", 3) 

215 self.assertEqual(len(schema.get_apdb_columns(ExtraTables.DiaObjectChunks)), 3) 

216 self._assertTable(schema.get_table(ExtraTables.DiaSourceChunks), "DiaSourceChunks", 2) 

217 self.assertEqual(len(schema.get_apdb_columns(ExtraTables.DiaSourceChunks)), 2) 

218 self._assertTable(schema.get_table(ExtraTables.DiaForcedSourceChunks), "DiaForcedSourceChunks", 4) 

219 self.assertEqual(len(schema.get_apdb_columns(ExtraTables.DiaForcedSourceChunks)), 4) 

220 

221 def test_makeSchema_nometa(self) -> None: 

222 """Make schema using old yaml file without metadata table.""" 

223 with update_schema_yaml(TEST_SCHEMA, drop_metadata=True) as schema_file: 

224 engine = create_engine("sqlite://") 

225 self.addCleanup(engine.dispose) 

226 schema = ApdbSqlSchema( 

227 table_schema=ApdbSchema(schema_file, TEST_SCHEMA_SSO), 

228 engine=engine, 

229 dia_object_index="baseline", 

230 htm_index_column="pixelId", 

231 ) 

232 schema.makeSchema(drop=True) 

233 with self.assertRaisesRegex(ValueError, "Table type ApdbTables.metadata does not exist"): 

234 schema.get_table(ApdbTables.metadata) 

235 

236 # Also check the case when database is missing metadata table but 

237 # YAML schema has it. 

238 schema = ApdbSqlSchema( 

239 table_schema=ApdbSchema(TEST_SCHEMA, TEST_SCHEMA_SSO), 

240 engine=engine, 

241 dia_object_index="baseline", 

242 htm_index_column="pixelId", 

243 ) 

244 with self.assertRaisesRegex(ValueError, "Table type ApdbTables.metadata does not exist"): 

245 schema.get_table(ApdbTables.metadata) 

246 

247 

248class MyMemoryTestCase(lsst.utils.tests.MemoryTestCase): 

249 """Run file leak tests.""" 

250 

251 

252def setup_module(module: Any) -> None: 

253 """Configure pytest.""" 

254 lsst.utils.tests.init() 

255 

256 

257if __name__ == "__main__": 

258 lsst.utils.tests.init() 

259 unittest.main()