Coverage for tests / test_cassandra_queries.py: 15%

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 

22import unittest 

23 

24from lsst.dax.apdb.cassandra.queries import Column, ColumnExpr, Delete, Insert, QExpr, Select, Update 

25 

26 

27class QExprTestCase(unittest.TestCase): 

28 """A test case for QExpr class.""" 

29 

30 def test_basic(self) -> None: 

31 """Test construction of the class.""" 

32 clause = QExpr("x = 100") 

33 self.assertEqual(clause.expression, "x = 100") 

34 self.assertEqual(clause.parameters, ()) 

35 self.assertFalse(clause.can_prepare) 

36 

37 clause = QExpr("x = {}", (100,)) 

38 self.assertEqual(clause.expression, "x = {}") 

39 self.assertEqual(clause.parameters, (100,)) 

40 self.assertTrue(clause.can_prepare) 

41 

42 clause = QExpr("x = 100", can_prepare=False) 

43 self.assertEqual(clause.expression, "x = 100") 

44 self.assertEqual(clause.parameters, ()) 

45 self.assertFalse(clause.can_prepare) 

46 

47 def test_error(self) -> None: 

48 """Test for exceptions.""" 

49 with self.assertRaisesRegex(ValueError, "Number of placeholders .* does not match"): 

50 QExpr("x = 100", (1,)) 

51 with self.assertRaisesRegex(ValueError, "Number of placeholders .* does not match"): 

52 QExpr("x = {} and y = {}", (1,)) 

53 

54 def test_combine(self) -> None: 

55 """Test combination of clauses.""" 

56 clause1 = QExpr("x = {}", (10,)) 

57 clause2 = QExpr("y = {}", (100,)) 

58 clause3 = QExpr("z = {}", (1000,), can_prepare=False) 

59 

60 clause = clause1 & clause2 

61 self.assertEqual(clause.expression, "x = {} AND y = {}") 

62 self.assertEqual(clause.parameters, (10, 100)) 

63 self.assertTrue(clause.can_prepare) 

64 

65 clause &= clause3 

66 self.assertEqual(clause.expression, "x = {} AND y = {} AND z = {}") 

67 self.assertEqual(clause.parameters, (10, 100, 1000)) 

68 self.assertFalse(clause.can_prepare) 

69 

70 def test_combine_products(self) -> None: 

71 """Test combination of clauses.""" 

72 clauses1 = [QExpr("x = {}", (10,)), QExpr("x = {}", (20,))] 

73 clauses2 = [QExpr("y = {}", (100,)), QExpr("y = {}", (200,))] 

74 extra = QExpr("z = 1000") # can_prepare will be False 

75 

76 clauses = list(QExpr.combine(clauses1, clauses2)) 

77 self.assertEqual(len(clauses), 4) 

78 self.assertEqual(clauses[0].expression, "x = {} AND y = {}") 

79 self.assertEqual(clauses[0].parameters, (10, 100)) 

80 self.assertEqual(clauses[1].expression, "x = {} AND y = {}") 

81 self.assertEqual(clauses[1].parameters, (10, 200)) 

82 self.assertEqual(clauses[2].expression, "x = {} AND y = {}") 

83 self.assertEqual(clauses[2].parameters, (20, 100)) 

84 self.assertEqual(clauses[3].expression, "x = {} AND y = {}") 

85 self.assertEqual(clauses[3].parameters, (20, 200)) 

86 self.assertTrue(clauses[3].can_prepare) 

87 

88 clauses = list(QExpr.combine(clauses1, [], extra=extra)) 

89 self.assertEqual(len(clauses), 2) 

90 self.assertEqual(clauses[0].expression, "x = {} AND z = 1000") 

91 self.assertEqual(clauses[0].parameters, (10,)) 

92 self.assertFalse(clauses[1].can_prepare) 

93 self.assertEqual(clauses[1].expression, "x = {} AND z = 1000") 

94 self.assertEqual(clauses[1].parameters, (20,)) 

95 self.assertFalse(clauses[1].can_prepare) 

96 

97 

98class ColumnTestCase(unittest.TestCase): 

99 """A test case for Column class.""" 

100 

101 def test_basic(self) -> None: 

102 """Test simple construction.""" 

103 c = Column("name") 

104 self.assertEqual(str(c), "name") 

105 

106 c = Column("Name") 

107 self.assertEqual(str(c), '"Name"') 

108 

109 def test_cmp(self) -> None: 

110 """Test comparisons.""" 

111 c1 = Column("name") 

112 c2 = Column("Instrument") 

113 self.assertEqual(c1 == 100, QExpr("name = {}", [100])) 

114 self.assertEqual("🔭" == c2, QExpr('"Instrument" = {}', ["🔭"])) 

115 self.assertEqual(c1 != 100, QExpr("name != {}", [100])) 

116 self.assertEqual("🔭" != c2, QExpr('"Instrument" != {}', ["🔭"])) 

117 self.assertEqual(c1 < 100, QExpr("name < {}", [100])) 

118 self.assertEqual("🔭" < c2, QExpr('"Instrument" > {}', ["🔭"])) 

119 self.assertEqual(c1 <= 100, QExpr("name <= {}", [100])) 

120 self.assertEqual("🔭" <= c2, QExpr('"Instrument" >= {}', ["🔭"])) 

121 self.assertEqual(c1 > 100, QExpr("name > {}", [100])) 

122 self.assertEqual("🔭" > c2, QExpr('"Instrument" < {}', ["🔭"])) 

123 self.assertEqual(c1 >= 100, QExpr("name >= {}", [100])) 

124 self.assertEqual("🔭" >= c2, QExpr('"Instrument" <= {}', ["🔭"])) 

125 

126 def test_in(self) -> None: 

127 """Test IN operator.""" 

128 c = Column("name") 

129 expr = c.in_([1, 2, 3]) 

130 self.assertEqual(expr, QExpr("name IN ({},{},{})", (1, 2, 3))) 

131 

132 

133class SelectQueryTestCase(unittest.TestCase): 

134 """A test case for Select class.""" 

135 

136 def test_basic(self) -> None: 

137 """Test simple construction.""" 

138 query = Select("keyspace", "table", ["*"]) 

139 self.assertEqual(str(query), "SELECT * FROM keyspace.table") 

140 

141 query = Select("Keyspace", "Table", ["A", "b", "c"]) 

142 self.assertEqual(str(query), 'SELECT "A",b,c FROM "Keyspace"."Table"') 

143 

144 query = Select("keyspace", "table", [ColumnExpr("COUNT(*)")], extra_clause="ALLOW FILTERING") 

145 self.assertEqual(str(query), "SELECT COUNT(*) FROM keyspace.table ALLOW FILTERING") 

146 

147 def test_where(self) -> None: 

148 """Test WHERE clause.""" 

149 query = Select("keyspace", "table", ["*"], extra_clause="LIMIT 1") 

150 query = query.where("x = {}", (10,)) 

151 self.assertEqual(str(query), "SELECT * FROM keyspace.table WHERE x = {} LIMIT 1") 

152 self.assertEqual(query.parameters, (10,)) 

153 query = query.where("y IN ({*})", [100, 101]) 

154 self.assertEqual(str(query), "SELECT * FROM keyspace.table WHERE x = {} AND y IN ({},{}) LIMIT 1") 

155 self.assertEqual(query.parameters, (10, 100, 101)) 

156 query = query.where("z = {}", [1000], can_prepare=False) 

157 self.assertEqual( 

158 str(query), "SELECT * FROM keyspace.table WHERE x = {} AND y IN ({},{}) AND z = {} LIMIT 1" 

159 ) 

160 self.assertEqual(query.parameters, (10, 100, 101, 1000)) 

161 

162 def test_can_prepare(self) -> None: 

163 """Test can_prepare handling.""" 

164 query = Select("keyspace", "table", ["*"]) 

165 query = query.where("x = {}", (10,)) 

166 self.assertTrue(query.can_prepare) 

167 

168 query = query.where("y = {}", (10,), can_prepare=False) 

169 self.assertFalse(query.can_prepare) 

170 

171 query = Select("keyspace", "table", ["*"], can_prepare=False) 

172 query = query.where("x = {}", (10,)) 

173 self.assertFalse(query.can_prepare) 

174 

175 def test_render(self) -> None: 

176 """Test render() method.""" 

177 query = Select("keyspace", "table", ["*"], where_clause=QExpr("x = {} AND y = {}", (10, 100))) 

178 self.assertEqual(query.render(), "SELECT * FROM keyspace.table WHERE x = {} AND y = {}") 

179 self.assertEqual(query.render("?"), "SELECT * FROM keyspace.table WHERE x = ? AND y = ?") 

180 self.assertEqual(query.render("%s"), "SELECT * FROM keyspace.table WHERE x = %s AND y = %s") 

181 

182 def test_errors(self) -> None: 

183 """Test exceptions.""" 

184 query = Select("keyspace", "table", ["*"]) 

185 with self.assertRaisesRegex(TypeError, "Unexpected arguments"): 

186 query.where() # type: ignore[call-overload] 

187 with self.assertRaisesRegex(TypeError, "Unexpected keyword arguments"): 

188 query.where("x = {}", (10,), cannot_prepare=False) # type: ignore[call-overload] 

189 with self.assertRaisesRegex(TypeError, "Unexpected arguments"): 

190 query.where("x = {}", 10, can_prepare=False) # type: ignore[call-overload] 

191 with self.assertRaisesRegex(TypeError, "Unexpected arguments"): 

192 query.where("x = {}", 10, 20) # type: ignore[call-overload] 

193 

194 

195class InsertQueryTestCase(unittest.TestCase): 

196 """A test case for Insert class.""" 

197 

198 def test_basic(self) -> None: 

199 """Test simple construction.""" 

200 query = Insert("keyspace", "table", ["x", "y"]) 

201 self.assertEqual(str(query), "INSERT INTO keyspace.table (x,y) VALUES ({},{})") 

202 query = Insert("Keyspace", "Table", ["X", "Y"]) 

203 self.assertEqual(str(query), 'INSERT INTO "Keyspace"."Table" ("X","Y") VALUES ({},{})') 

204 

205 def test_can_prepare(self) -> None: 

206 """Test can_prepare handling.""" 

207 query = Insert("keyspace", "table", ["x", "y"]) 

208 self.assertTrue(query.can_prepare) 

209 

210 query = Insert("keyspace", "table", ["x", "y"], can_prepare=False) 

211 self.assertFalse(query.can_prepare) 

212 

213 def test_render(self) -> None: 

214 """Test render() method.""" 

215 query = Insert("keyspace", "table", ["x", "y"]) 

216 self.assertEqual(query.render(), "INSERT INTO keyspace.table (x,y) VALUES ({},{})") 

217 self.assertEqual(query.render("?"), "INSERT INTO keyspace.table (x,y) VALUES (?,?)") 

218 self.assertEqual(query.render("%s"), "INSERT INTO keyspace.table (x,y) VALUES (%s,%s)") 

219 

220 

221class DeleteQueryTestCase(unittest.TestCase): 

222 """A test case for Delete class.""" 

223 

224 def test_basic(self) -> None: 

225 """Test simple construction.""" 

226 query = Delete("keyspace", "table") 

227 query = query.where("x = {}", (10,)) 

228 self.assertEqual(str(query), "DELETE FROM keyspace.table WHERE x = {}") 

229 self.assertEqual(query.parameters, (10,)) 

230 

231 query = Delete("Keyspace", "Table") 

232 query = query.where('"Y" = {}', (10,)) 

233 self.assertEqual(str(query), 'DELETE FROM "Keyspace"."Table" WHERE "Y" = {}') 

234 self.assertEqual(query.parameters, (10,)) 

235 

236 def test_can_prepare(self) -> None: 

237 """Test can_prepare handling.""" 

238 query = Delete("keyspace", "table") 

239 query = query.where("x = {}", (10,)) 

240 self.assertTrue(query.can_prepare) 

241 

242 query = query.where("y = {}", (10,), can_prepare=False) 

243 self.assertFalse(query.can_prepare) 

244 

245 query = Delete("keyspace", "table", can_prepare=False) 

246 query = query.where("x = {}", (10,)) 

247 self.assertFalse(query.can_prepare) 

248 

249 def test_render(self) -> None: 

250 """Test render() method.""" 

251 query = Delete("keyspace", "table", where_clause=QExpr("x = {} AND y = {}", (10, 100))) 

252 self.assertEqual(query.render(), "DELETE FROM keyspace.table WHERE x = {} AND y = {}") 

253 self.assertEqual(query.render("?"), "DELETE FROM keyspace.table WHERE x = ? AND y = ?") 

254 self.assertEqual(query.render("%s"), "DELETE FROM keyspace.table WHERE x = %s AND y = %s") 

255 

256 def test_errors(self) -> None: 

257 """Test exceptions.""" 

258 query = Delete("keyspace", "table") 

259 with self.assertRaisesRegex(RuntimeError, "DELETE statement without WHERE clause"): 

260 query.render() 

261 with self.assertRaisesRegex(TypeError, "Unexpected arguments"): 

262 query.where() # type: ignore[call-overload] 

263 

264 

265class UpdateQueryTestCase(unittest.TestCase): 

266 """A test case for Update class.""" 

267 

268 def test_basic(self) -> None: 

269 """Test simple construction.""" 

270 query = Update("keyspace", "table") 

271 query = query.where("x = {}", (10,)) 

272 query = query.values(QExpr("x = {}", (100,))) 

273 self.assertEqual(str(query), "UPDATE keyspace.table SET x = {} WHERE x = {}") 

274 self.assertEqual(query.parameters, (100, 10)) 

275 

276 query = Update("Keyspace", "Table") 

277 query = query.values(Column("Y").update(20), Column("Z").update(30)) 

278 query = query.where('"Y" = {}', (10,)) 

279 self.assertEqual(str(query), 'UPDATE "Keyspace"."Table" SET "Y" = {}, "Z" = {} WHERE "Y" = {}') 

280 self.assertEqual(query.parameters, (20, 30, 10)) 

281 

282 def test_can_prepare(self) -> None: 

283 """Test can_prepare handling.""" 

284 query = Update("keyspace", "table") 

285 query = query.where("x = {}", (10,)) 

286 query = query.values(Column("x").update(100)) 

287 self.assertTrue(query.can_prepare) 

288 

289 query = query.where("y = {}", (10,), can_prepare=False) 

290 self.assertFalse(query.can_prepare) 

291 

292 query = Update("keyspace", "table", can_prepare=False) 

293 query = query.where("x = {}", (10,)) 

294 query = query.values(Column("x").update(100)) 

295 self.assertFalse(query.can_prepare) 

296 

297 def test_render(self) -> None: 

298 """Test render() method.""" 

299 query = Update("keyspace", "table", where_clause=QExpr("x = {} AND y = {}", (10, 200))) 

300 query = query.values(Column("x").update(100)) 

301 query = query.values(Column("y").update(20)) 

302 self.assertEqual(query.parameters, (100, 20, 10, 200)) 

303 self.assertEqual(query.render(), "UPDATE keyspace.table SET x = {}, y = {} WHERE x = {} AND y = {}") 

304 self.assertEqual(query.render("?"), "UPDATE keyspace.table SET x = ?, y = ? WHERE x = ? AND y = ?") 

305 self.assertEqual( 

306 query.render("%s"), "UPDATE keyspace.table SET x = %s, y = %s WHERE x = %s AND y = %s" 

307 ) 

308 

309 def test_errors(self) -> None: 

310 """Test exceptions.""" 

311 query = Update("keyspace", "table") 

312 with self.assertRaisesRegex(RuntimeError, "UPDATE statement without WHERE clause"): 

313 query.render() 

314 query = query.where("x = {}", (10,)) 

315 with self.assertRaisesRegex(RuntimeError, "UPDATE statement without SET clause"): 

316 query.render() 

317 

318 

319if __name__ == "__main__": 

320 unittest.main()