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 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: 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 table_schema=ApdbSchema(TEST_SCHEMA, TEST_SCHEMA_SSO), 

83 engine=engine, 

84 dia_object_index="baseline", 

85 htm_index_column="pixelId", 

86 ) 

87 schema.makeSchema() 

88 

89 for apdb_table in ApdbTables: 

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

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

92 table = schema.get_table(apdb_table) 

93 continue 

94 

95 table = schema.get_table(apdb_table) 

96 

97 column_count = self.table_column_count[apdb_table] 

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

99 # DiaObject and DiaSource tables add pixelId column. 

100 column_count += 1 

101 

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

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

104 if apdb_table is ApdbTables.DiaObject: 

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

106 

107 for table_enum in ExtraTables: 

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

109 schema.get_table(table_enum) 

110 

111 def test_makeSchema_prefix(self) -> None: 

112 """Create schema using prefix.""" 

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

114 schema = ApdbSqlSchema( 

115 table_schema=ApdbSchema(TEST_SCHEMA_COMBINED, ""), 

116 engine=engine, 

117 dia_object_index="baseline", 

118 htm_index_column="pixelId", 

119 prefix="Pfx", 

120 ) 

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

122 schema.makeSchema(drop=True) 

123 self._assertTable( 

124 schema.get_table(ApdbTables.DiaObject), 

125 "PfxDiaObject", 

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

127 ) 

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

129 schema.get_table(ApdbTables.DiaObjectLast) 

130 self._assertTable( 

131 schema.get_table(ApdbTables.DiaSource), 

132 "PfxDiaSource", 

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

134 ) 

135 self._assertTable( 

136 schema.get_table(ApdbTables.DiaForcedSource), 

137 "PfxDiaForcedSource", 

138 self.table_column_count[ApdbTables.DiaForcedSource], 

139 ) 

140 

141 def test_makeSchema_other_index(self) -> None: 

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

143 columns. 

144 """ 

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

146 schema = ApdbSqlSchema( 

147 table_schema=ApdbSchema(TEST_SCHEMA, TEST_SCHEMA_SSO), 

148 engine=engine, 

149 dia_object_index="pix_id_iov", 

150 htm_index_column="pixelId", 

151 ) 

152 schema.makeSchema(drop=True) 

153 table = schema.get_table(ApdbTables.DiaObject) 

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

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

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

157 schema.get_table(ApdbTables.DiaObjectLast) 

158 self._assertTable( 

159 schema.get_table(ApdbTables.DiaSource), 

160 "DiaSource", 

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

162 ) 

163 self._assertTable( 

164 schema.get_table(ApdbTables.DiaForcedSource), 

165 "DiaForcedSource", 

166 self.table_column_count[ApdbTables.DiaForcedSource], 

167 ) 

168 

169 def test_makeSchema_diaobjectlast(self) -> None: 

170 """Use DiaObjectLast table for DiaObject.""" 

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

172 schema = ApdbSqlSchema( 

173 table_schema=ApdbSchema(TEST_SCHEMA, TEST_SCHEMA_SSO), 

174 engine=engine, 

175 dia_object_index="last_object_table", 

176 htm_index_column="pixelId", 

177 ) 

178 schema.makeSchema(drop=True) 

179 table = schema.get_table(ApdbTables.DiaObject) 

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

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

182 table = schema.get_table(ApdbTables.DiaObjectLast) 

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

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

185 self._assertTable( 

186 schema.get_table(ApdbTables.DiaSource), 

187 "DiaSource", 

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

189 ) 

190 self._assertTable( 

191 schema.get_table(ApdbTables.DiaForcedSource), 

192 "DiaForcedSource", 

193 self.table_column_count[ApdbTables.DiaForcedSource], 

194 ) 

195 

196 def test_makeSchema_replica(self) -> None: 

197 """Add replica tables.""" 

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

199 schema = ApdbSqlSchema( 

200 table_schema=ApdbSchema(TEST_SCHEMA, TEST_SCHEMA_SSO), 

201 engine=engine, 

202 dia_object_index="last_object_table", 

203 htm_index_column="pixelId", 

204 enable_replica=True, 

205 ) 

206 schema.makeSchema(drop=True) 

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

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

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

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

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

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

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

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

215 

216 def test_makeSchema_nometa(self) -> None: 

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

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

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

220 schema = ApdbSqlSchema( 

221 table_schema=ApdbSchema(schema_file, TEST_SCHEMA_SSO), 

222 engine=engine, 

223 dia_object_index="baseline", 

224 htm_index_column="pixelId", 

225 ) 

226 schema.makeSchema(drop=True) 

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

228 schema.get_table(ApdbTables.metadata) 

229 

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

231 # YAML schema has it. 

232 schema = ApdbSqlSchema( 

233 table_schema=ApdbSchema(TEST_SCHEMA, TEST_SCHEMA_SSO), 

234 engine=engine, 

235 dia_object_index="baseline", 

236 htm_index_column="pixelId", 

237 ) 

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

239 schema.get_table(ApdbTables.metadata) 

240 

241 

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

243 """Run file leak tests.""" 

244 

245 

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

247 """Configure pytest.""" 

248 lsst.utils.tests.init() 

249 

250 

251if __name__ == "__main__": 

252 lsst.utils.tests.init() 

253 unittest.main()