Coverage for tests/test_apdbSqlSchema.py: 20%

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 

36TEST_SCHEMA = os.path.join(os.path.abspath(os.path.dirname(__file__)), "config/schema.yaml") 

37 

38 

39class ApdbSchemaTestCase(unittest.TestCase): 

40 """Test case for ApdbSqlSchema class. 

41 

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

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

44 """ 

45 

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

47 table_column_count = { 

48 ApdbTables.DiaObject: 7, 

49 ApdbTables.DiaObjectLast: 5, 

50 ApdbTables.DiaSource: 12, 

51 ApdbTables.DiaForcedSource: 8, 

52 ApdbTables.DiaObject_To_Object_Match: 3, 

53 ApdbTables.SSObject: 3, 

54 ApdbTables.SSSource: 4, 

55 ApdbTables.metadata: 2, 

56 } 

57 

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

59 """Validate tables schema. 

60 

61 Parameters 

62 ---------- 

63 table : `sqlalchemy.Table` 

64 name : `str` 

65 Expected table name 

66 ncol : `int` 

67 Expected number of columns 

68 """ 

69 self.assertIsNotNone(table) 

70 self.assertEqual(table.name, name) 

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

72 

73 def test_makeSchema_default(self) -> None: 

74 """Test for creating schema.""" 

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

76 

77 # create standard (baseline) schema 

78 schema = ApdbSqlSchema( 

79 engine=engine, dia_object_index="baseline", htm_index_column="pixelId", schema_file=TEST_SCHEMA 

80 ) 

81 schema.makeSchema() 

82 

83 for apdb_table in ApdbTables: 

84 if apdb_table in (ApdbTables.DiaObjectLast, ApdbTables.SSSource): 

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

86 table = schema.get_table(apdb_table) 

87 continue 

88 

89 table = schema.get_table(apdb_table) 

90 

91 column_count = self.table_column_count[apdb_table] 

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

93 # DiaObject and DiaSource tables add pixelId column. 

94 column_count += 1 

95 

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

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

98 if apdb_table is ApdbTables.DiaObject: 

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

100 

101 for table_enum in ExtraTables: 

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

103 schema.get_table(table_enum) 

104 

105 def test_makeSchema_prefix(self) -> None: 

106 """Create schema using prefix.""" 

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

108 schema = ApdbSqlSchema( 

109 engine=engine, 

110 dia_object_index="baseline", 

111 htm_index_column="pixelId", 

112 schema_file=TEST_SCHEMA, 

113 prefix="Pfx", 

114 ) 

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

116 schema.makeSchema(drop=True) 

117 self._assertTable( 

118 schema.get_table(ApdbTables.DiaObject), 

119 "PfxDiaObject", 

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

121 ) 

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

123 schema.get_table(ApdbTables.DiaObjectLast) 

124 self._assertTable( 

125 schema.get_table(ApdbTables.DiaSource), 

126 "PfxDiaSource", 

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

128 ) 

129 self._assertTable( 

130 schema.get_table(ApdbTables.DiaForcedSource), 

131 "PfxDiaForcedSource", 

132 self.table_column_count[ApdbTables.DiaForcedSource], 

133 ) 

134 

135 def test_makeSchema_other_index(self) -> None: 

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

137 columns. 

138 """ 

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

140 schema = ApdbSqlSchema( 

141 engine=engine, dia_object_index="pix_id_iov", htm_index_column="pixelId", schema_file=TEST_SCHEMA 

142 ) 

143 schema.makeSchema(drop=True) 

144 table = schema.get_table(ApdbTables.DiaObject) 

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

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

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

148 schema.get_table(ApdbTables.DiaObjectLast) 

149 self._assertTable( 

150 schema.get_table(ApdbTables.DiaSource), 

151 "DiaSource", 

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

153 ) 

154 self._assertTable( 

155 schema.get_table(ApdbTables.DiaForcedSource), 

156 "DiaForcedSource", 

157 self.table_column_count[ApdbTables.DiaForcedSource], 

158 ) 

159 

160 def test_makeSchema_diaobjectlast(self) -> None: 

161 """Use DiaObjectLast table for DiaObject.""" 

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

163 schema = ApdbSqlSchema( 

164 engine=engine, 

165 dia_object_index="last_object_table", 

166 htm_index_column="pixelId", 

167 schema_file=TEST_SCHEMA, 

168 ) 

169 schema.makeSchema(drop=True) 

170 table = schema.get_table(ApdbTables.DiaObject) 

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

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

173 table = schema.get_table(ApdbTables.DiaObjectLast) 

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

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

176 self._assertTable( 

177 schema.get_table(ApdbTables.DiaSource), 

178 "DiaSource", 

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

180 ) 

181 self._assertTable( 

182 schema.get_table(ApdbTables.DiaForcedSource), 

183 "DiaForcedSource", 

184 self.table_column_count[ApdbTables.DiaForcedSource], 

185 ) 

186 

187 def test_makeSchema_replica(self) -> None: 

188 """Add replica tables.""" 

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

190 schema = ApdbSqlSchema( 

191 engine=engine, 

192 dia_object_index="last_object_table", 

193 htm_index_column="pixelId", 

194 schema_file=TEST_SCHEMA, 

195 enable_replica=True, 

196 ) 

197 schema.makeSchema(drop=True) 

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

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

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

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

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

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

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

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

206 

207 def test_makeSchema_nometa(self) -> None: 

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

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

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

211 schema = ApdbSqlSchema( 

212 engine=engine, 

213 dia_object_index="baseline", 

214 htm_index_column="pixelId", 

215 schema_file=schema_file, 

216 ) 

217 schema.makeSchema(drop=True) 

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

219 schema.get_table(ApdbTables.metadata) 

220 

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

222 # YAML schema has it. 

223 schema = ApdbSqlSchema( 

224 engine=engine, 

225 dia_object_index="baseline", 

226 htm_index_column="pixelId", 

227 schema_file=TEST_SCHEMA, 

228 ) 

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

230 schema.get_table(ApdbTables.metadata) 

231 

232 

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

234 """Run file leak tests.""" 

235 

236 

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

238 """Configure pytest.""" 

239 lsst.utils.tests.init() 

240 

241 

242if __name__ == "__main__": 

243 lsst.utils.tests.init() 

244 unittest.main()