Coverage for tests / test_apdbSqlSchema.py: 23%

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

52 ApdbTables.DiaObjectLast: 5, 

53 ApdbTables.DiaSource: 12, 

54 ApdbTables.DiaForcedSource: 8, 

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 

80 # create standard (baseline) schema 

81 schema = ApdbSqlSchema( 

82 engine=engine, 

83 dia_object_index="baseline", 

84 htm_index_column="pixelId", 

85 schema_file=TEST_SCHEMA, 

86 ss_schema_file=TEST_SCHEMA_SSO, 

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 schema = ApdbSqlSchema( 

116 engine=engine, 

117 dia_object_index="baseline", 

118 htm_index_column="pixelId", 

119 schema_file=TEST_SCHEMA_COMBINED, 

120 ss_schema_file="", 

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 schema = ApdbSqlSchema( 

149 engine=engine, 

150 dia_object_index="pix_id_iov", 

151 htm_index_column="pixelId", 

152 schema_file=TEST_SCHEMA, 

153 ss_schema_file=TEST_SCHEMA_SSO, 

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 schema = ApdbSqlSchema( 

176 engine=engine, 

177 dia_object_index="last_object_table", 

178 htm_index_column="pixelId", 

179 schema_file=TEST_SCHEMA, 

180 ss_schema_file=TEST_SCHEMA_SSO, 

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 schema = ApdbSqlSchema( 

204 engine=engine, 

205 dia_object_index="last_object_table", 

206 htm_index_column="pixelId", 

207 schema_file=TEST_SCHEMA, 

208 ss_schema_file=TEST_SCHEMA_SSO, 

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 schema = ApdbSqlSchema( 

226 engine=engine, 

227 dia_object_index="baseline", 

228 htm_index_column="pixelId", 

229 schema_file=schema_file, 

230 ss_schema_file=TEST_SCHEMA_SSO, 

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 engine=engine, 

240 dia_object_index="baseline", 

241 htm_index_column="pixelId", 

242 schema_file=TEST_SCHEMA, 

243 ss_schema_file=TEST_SCHEMA_SSO, 

244 ) 

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

246 schema.get_table(ApdbTables.metadata) 

247 

248 

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

250 """Run file leak tests.""" 

251 

252 

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

254 """Configure pytest.""" 

255 lsst.utils.tests.init() 

256 

257 

258if __name__ == "__main__": 

259 lsst.utils.tests.init() 

260 unittest.main()