Coverage for tests / test_apdbCassandraAdmin.py: 34%

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 `ApdbCassandraAdmin` class. 

23 

24Notes 

25----- 

26For now this test can only run against actual Cassandra cluster, to specify 

27cluster location use ``DAX_APDB_TEST_CASSANDRA_CLUSTER`` environment variable, 

28e.g.: 

29 

30 export DAX_APDB_TEST_CASSANDRA_CLUSTER=cassandra.example.com 

31 pytest tests/test_apdbCassandra.py 

32 

33Individual tests create and destroy unique keyspaces in the cluster, there is 

34no need to pre-create a keyspace with predefined name. 

35""" 

36 

37import logging 

38import os 

39import unittest 

40from typing import Any, cast 

41 

42import astropy.time 

43 

44from lsst.dax.apdb import Apdb, ApdbConfig 

45from lsst.dax.apdb.cassandra import ApdbCassandra 

46from lsst.dax.apdb.cassandra.apdbCassandraAdmin import ApdbCassandraAdmin 

47from lsst.dax.apdb.cassandra.partitioner import Partitioner 

48from lsst.dax.apdb.tests import ApdbAdminTest, cassandra_mixin 

49 

50TEST_SCHEMA = os.path.join(os.path.abspath(os.path.dirname(__file__)), "config/schema-apdb+sso.yaml") 

51 

52logging.basicConfig(level=logging.INFO) 

53 

54 

55class ApdbCassandraAdminTestCase(cassandra_mixin.ApdbCassandraMixin, ApdbAdminTest, unittest.TestCase): 

56 """A test case for ApdbCassandraAdmin class.""" 

57 

58 schema_path = TEST_SCHEMA 

59 time_partition_tables = False 

60 time_partition_start: str | None = None 

61 time_partition_end: str | None = None 

62 

63 def make_instance(self, **kwargs: Any) -> ApdbConfig: 

64 """Make config class instance used in all tests.""" 

65 kw: dict[str, Any] = { 

66 "hosts": (self.cluster_host,), 

67 "keyspace": self.keyspace, 

68 "schema_file": TEST_SCHEMA, 

69 "time_partition_tables": self.time_partition_tables, 

70 } 

71 if self.time_partition_start: 

72 kw["time_partition_start"] = self.time_partition_start 

73 if self.time_partition_end: 

74 kw["time_partition_end"] = self.time_partition_end 

75 kw.update(kwargs) 

76 return ApdbCassandra.init_database(**kw) 

77 

78 def make_admin(self) -> ApdbCassandraAdmin: 

79 """Make ApdbCassandraAdmin instance.""" 

80 config = self.make_instance() 

81 apdb = cast(ApdbCassandra, Apdb.from_config(config)) 

82 return apdb.admin 

83 

84 def test_extend_time_partitions(self) -> None: 

85 """Test extend_time_partitions() method.""" 

86 admin = self.make_admin() 

87 

88 self._test_extend_time_partitions(admin) 

89 

90 def _test_extend_time_partitions(self, admin: ApdbCassandraAdmin) -> None: 

91 """Implement test_extend_time_partitions, to be specialized in 

92 subclasses. 

93 """ 

94 with self.assertRaisesRegex(TypeError, "does not use time-partitioned tables"): 

95 admin.time_partitions() 

96 with self.assertRaisesRegex(TypeError, "does not use time-partitioned tables"): 

97 admin.extend_time_partitions(astropy.time.Time("2020-01-01T00:00:00Z", format="isot")) 

98 

99 def test_delete_time_partitions(self) -> None: 

100 """Test delete_time_partitions() method.""" 

101 admin = self.make_admin() 

102 

103 self._test_delete_time_partitions(admin) 

104 

105 def _test_delete_time_partitions(self, admin: ApdbCassandraAdmin) -> None: 

106 """Implement test_delete_time_partitions, to be specialized in 

107 subclasses. 

108 """ 

109 with self.assertRaisesRegex(TypeError, "does not use time-partitioned tables"): 

110 admin.delete_time_partitions(astropy.time.Time("2020-01-01T00:00:00Z", format="isot")) 

111 

112 

113class ApdbCassandraAdminTimePartitionedTestCase(ApdbCassandraAdminTestCase): 

114 """A test case for ApdbCassandraAdmin class with per-month tables.""" 

115 

116 time_partition_tables = True 

117 time_partition_start = "2025-01-01T00:00:00" 

118 time_partition_end = "2026-01-01T00:00:00" 

119 

120 def _test_extend_time_partitions(self, admin: ApdbCassandraAdmin) -> None: 

121 time_part = admin.time_partitions() 

122 self.assertEqual((time_part.start, time_part.end), (669, 681)) 

123 

124 # Test too long range first. 

125 with self.assertRaisesRegex(ValueError, "Extension exceeds limit"): 

126 admin.extend_time_partitions(astropy.time.Time("2027-02-02T00:00:00Z", format="isot")) 

127 with self.assertRaisesRegex(ValueError, "Extension exceeds limit"): 

128 admin.extend_time_partitions( 

129 astropy.time.Time("2023-11-25T00:00:00Z", format="isot"), forward=False 

130 ) 

131 

132 # Nothing to do. 

133 self.assertFalse( 

134 admin.extend_time_partitions(astropy.time.Time("2024-01-01T00:00:00Z", format="isot")) 

135 ) 

136 self.assertFalse( 

137 admin.extend_time_partitions( 

138 astropy.time.Time("2027-01-01T00:00:00Z", format="isot"), forward=False 

139 ) 

140 ) 

141 time_part = admin.time_partitions() 

142 self.assertEqual((time_part.start, time_part.end), (669, 681)) 

143 

144 self.assertTrue( 

145 admin.extend_time_partitions(astropy.time.Time("2026-12-01T00:00:00Z", format="isot")) 

146 ) 

147 time_part = admin.time_partitions() 

148 self.assertEqual((time_part.start, time_part.end), (669, 692)) 

149 self.assertTrue( 

150 admin.extend_time_partitions( 

151 astropy.time.Time("2024-02-01T00:00:00Z", format="isot"), forward=False 

152 ) 

153 ) 

154 time_part = admin.time_partitions() 

155 self.assertEqual((time_part.start, time_part.end), (658, 692)) 

156 

157 def _no_confirm(self, *, partitions: list[int], tables: list[str], partitioner: Partitioner) -> bool: 

158 return False 

159 

160 def _confirm(self, *, partitions: list[int], tables: list[str], partitioner: Partitioner) -> bool: 

161 return True 

162 

163 def _test_delete_time_partitions(self, admin: ApdbCassandraAdmin) -> None: 

164 time_part = admin.time_partitions() 

165 self.assertEqual((time_part.start, time_part.end), (669, 681)) 

166 

167 # Should not delete anyhting. 

168 partitions = admin.delete_time_partitions( 

169 astropy.time.Time(self.time_partition_start, format="isot", scale="tai") 

170 ) 

171 self.assertEqual(partitions, []) 

172 partitions = admin.delete_time_partitions( 

173 astropy.time.Time(self.time_partition_end, format="isot", scale="tai"), after=True 

174 ) 

175 self.assertEqual(partitions, []) 

176 partitions = admin.delete_time_partitions(astropy.time.Time("2020-01-01", format="isot", scale="tai")) 

177 self.assertEqual(partitions, []) 

178 partitions = admin.delete_time_partitions( 

179 astropy.time.Time("2030-01-01", format="isot", scale="tai"), after=True 

180 ) 

181 self.assertEqual(partitions, []) 

182 

183 # Deletes one partiion 

184 partitions = admin.delete_time_partitions(astropy.time.Time("2025-01-31", format="isot", scale="tai")) 

185 self.assertEqual(partitions, [669]) 

186 time_part = admin.time_partitions() 

187 self.assertEqual((time_part.start, time_part.end), (670, 681)) 

188 

189 # Not confirmed. 

190 partitions = admin.delete_time_partitions( 

191 astropy.time.Time("2025-03-31", format="isot", scale="tai"), confirm=self._no_confirm 

192 ) 

193 self.assertEqual(partitions, []) 

194 

195 # Confirmed. 

196 partitions = admin.delete_time_partitions( 

197 astropy.time.Time("2025-11-02", format="isot", scale="tai"), after=True, confirm=self._confirm 

198 ) 

199 self.assertEqual(partitions, [680, 681]) 

200 time_part = admin.time_partitions() 

201 self.assertEqual((time_part.start, time_part.end), (670, 679)) 

202 

203 with self.assertRaisesRegex(ValueError, "Cannot delete all partitions"): 

204 admin.delete_time_partitions(astropy.time.Time("2030-01-01", format="isot", scale="tai")) 

205 

206 

207class ApdbCassandraAdminNoDiaObjectTestCase(ApdbCassandraAdminTimePartitionedTestCase): 

208 """A test case for ApdbCassandraAdmin with replica table and no DiaObject 

209 table. 

210 """ 

211 

212 def make_instance(self, **kwargs: Any) -> ApdbConfig: 

213 return super().make_instance(enable_replica=True, replica_skips_diaobjects=True) 

214 

215 

216if __name__ == "__main__": 

217 unittest.main()