Coverage for tests / test_sqlite.py: 37%

1# This file is part of daf_butler. 

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 software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

15# This program is free software: you can redistribute it and/or modify 

16# it under the terms of the GNU General Public License as published by 

17# the Free Software Foundation, either version 3 of the License, or 

18# (at your option) any later version. 

19# 

20# This program is distributed in the hope that it will be useful, 

21# but WITHOUT ANY WARRANTY; without even the implied warranty of 

22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

23# GNU General Public License for more details. 

24# 

25# You should have received a copy of the GNU General Public License 

26# along with this program. If not, see <http://www.gnu.org/licenses/>. 

27 

28import os 

29import os.path 

30import stat 

31import tempfile 

32import unittest 

33from contextlib import contextmanager 

34from typing import cast 

35 

36import sqlalchemy 

37 

38from lsst.daf.butler import Butler, Config, ddl 

39from lsst.daf.butler.registry import RegistryConfig, _RegistryFactory 

40from lsst.daf.butler.registry.databases.sqlite import SqliteDatabase 

41from lsst.daf.butler.registry.tests import DatabaseTests, RegistryTests 

42from lsst.daf.butler.tests import makeTestRepo 

43from lsst.daf.butler.tests.utils import create_populated_sqlite_registry, makeTestTempDir, removeTestTempDir 

44 

45TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

46 

47 

48@contextmanager 

49def removeWritePermission(filename): 

50 """Remove the write permission on a file.""" 

51 mode = os.stat(filename).st_mode 

52 try: 

53 os.chmod(filename, stat.S_IREAD) 

54 yield 

55 finally: 

56 os.chmod(filename, mode) 

57 

58 

59def isEmptyDatabaseActuallyWriteable(database: SqliteDatabase) -> bool: 

60 """Check whether we really can modify a database. 

61 

62 This intentionally allows any exception to be raised (not just 

63 `ReadOnlyDatabaseError`) to deal with cases where the file is read-only 

64 but the Database was initialized (incorrectly) with writeable=True. 

65 """ 

66 try: 

67 with database.declareStaticTables(create=True) as context: 

68 table = context.addTable( 

69 "a", ddl.TableSpec(fields=[ddl.FieldSpec("b", dtype=sqlalchemy.Integer, primaryKey=True)]) 

70 ) 

71 # Drop created table so that schema remains empty. 

72 database._metadata._metadata.drop_all(database._engine, tables=[table]) 

73 return True 

74 except Exception: 

75 return False 

76 

77 

78class SqliteFileDatabaseTestCase(unittest.TestCase, DatabaseTests): 

79 """Tests for `SqliteDatabase` using a standard file-based database.""" 

80 

81 def setUp(self): 

82 self.root = makeTestTempDir(TESTDIR) 

83 

84 def tearDown(self): 

85 removeTestTempDir(self.root) 

86 

87 def makeEmptyDatabase(self, origin: int = 0) -> SqliteDatabase: 

88 _, filename = tempfile.mkstemp(dir=self.root, suffix=".sqlite3") 

89 engine = SqliteDatabase.makeEngine(filename=filename) 

90 db = SqliteDatabase.fromEngine(engine=engine, origin=origin) 

91 self.addCleanup(db.dispose) 

92 return db 

93 

94 def getNewConnection(self, database: SqliteDatabase, *, writeable: bool) -> SqliteDatabase: 

95 engine = SqliteDatabase.makeEngine(filename=database.filename, writeable=writeable) 

96 db = SqliteDatabase.fromEngine(origin=database.origin, engine=engine, writeable=writeable) 

97 self.addCleanup(db.dispose) 

98 return db 

99 

100 @contextmanager 

101 def asReadOnly(self, database: SqliteDatabase) -> SqliteDatabase: 

102 with removeWritePermission(database.filename): 

103 yield self.getNewConnection(database, writeable=False) 

104 

105 def testConnection(self): 

106 """Test that different ways of connecting to a SQLite database 

107 are equivalent. 

108 """ 

109 _, filename = tempfile.mkstemp(dir=self.root, suffix=".sqlite3") 

110 # Create a read-write database by passing in the filename. 

111 rwFromFilename = SqliteDatabase.fromEngine(SqliteDatabase.makeEngine(filename=filename), origin=0) 

112 self.addCleanup(rwFromFilename.dispose) 

113 self.assertEqual(os.path.realpath(rwFromFilename.filename), os.path.realpath(filename)) 

114 self.assertEqual(rwFromFilename.origin, 0) 

115 self.assertTrue(rwFromFilename.isWriteable()) 

116 self.assertTrue(isEmptyDatabaseActuallyWriteable(rwFromFilename)) 

117 # Create a read-write database via a URI. 

118 rwFromUri = SqliteDatabase.fromUri(f"sqlite:///{filename}", origin=0) 

119 self.addCleanup(rwFromUri.dispose) 

120 self.assertEqual(os.path.realpath(rwFromUri.filename), os.path.realpath(filename)) 

121 self.assertEqual(rwFromUri.origin, 0) 

122 self.assertTrue(rwFromUri.isWriteable()) 

123 self.assertTrue(isEmptyDatabaseActuallyWriteable(rwFromUri)) 

124 # We don't support SQLite URIs inside SQLAlchemy URIs. 

125 with self.assertRaises(NotImplementedError): 

126 SqliteDatabase.makeEngine(uri=f"sqlite:///file:{filename}?uri=true") 

127 

128 # Test read-only connections against a read-only file. 

129 with removeWritePermission(filename): 

130 # Create a read-only database by passing in the filename. 

131 roFromFilename = SqliteDatabase.fromEngine( 

132 SqliteDatabase.makeEngine(filename=filename), origin=0, writeable=False 

133 ) 

134 self.addCleanup(roFromFilename.dispose) 

135 self.assertEqual(os.path.realpath(roFromFilename.filename), os.path.realpath(filename)) 

136 self.assertEqual(roFromFilename.origin, 0) 

137 self.assertFalse(roFromFilename.isWriteable()) 

138 self.assertFalse(isEmptyDatabaseActuallyWriteable(roFromFilename)) 

139 # Create a read-write database via a URI. 

140 roFromUri = SqliteDatabase.fromUri(f"sqlite:///{filename}", origin=0, writeable=False) 

141 self.addCleanup(roFromUri.dispose) 

142 self.assertEqual(os.path.realpath(roFromUri.filename), os.path.realpath(filename)) 

143 self.assertEqual(roFromUri.origin, 0) 

144 self.assertFalse(roFromUri.isWriteable()) 

145 self.assertFalse(isEmptyDatabaseActuallyWriteable(roFromUri)) 

146 

147 def testTransactionLocking(self): 

148 # This (inherited) test can't run on SQLite because of our use of an 

149 # aggressive locking strategy there. 

150 pass 

151 

152 

153class SqliteMemoryDatabaseTestCase(unittest.TestCase, DatabaseTests): 

154 """Tests for `SqliteDatabase` using an in-memory database.""" 

155 

156 def makeEmptyDatabase(self, origin: int = 0) -> SqliteDatabase: 

157 engine = SqliteDatabase.makeEngine(filename=None) 

158 db = SqliteDatabase.fromEngine(engine=engine, origin=origin) 

159 self.addCleanup(db.dispose) 

160 return db 

161 

162 def getNewConnection(self, database: SqliteDatabase, *, writeable: bool) -> SqliteDatabase: 

163 db = SqliteDatabase.fromEngine(origin=database.origin, engine=database._engine, writeable=writeable) 

164 self.addCleanup(db.dispose) 

165 return db 

166 

167 @contextmanager 

168 def asReadOnly(self, database: SqliteDatabase) -> SqliteDatabase: 

169 yield self.getNewConnection(database, writeable=False) 

170 

171 def testConnection(self): 

172 """Test that different ways of connecting to a SQLite database 

173 are equivalent. 

174 """ 

175 # Create an in-memory database by passing filename=None. 

176 memFromFilename = SqliteDatabase.fromEngine(SqliteDatabase.makeEngine(filename=None), origin=0) 

177 self.addCleanup(memFromFilename.dispose) 

178 self.assertIsNone(memFromFilename.filename) 

179 self.assertEqual(memFromFilename.origin, 0) 

180 self.assertTrue(memFromFilename.isWriteable()) 

181 self.assertTrue(isEmptyDatabaseActuallyWriteable(memFromFilename)) 

182 # Create an in-memory database via a URI. 

183 memFromUri = SqliteDatabase.fromUri("sqlite://", origin=0) 

184 self.addCleanup(memFromUri.dispose) 

185 self.assertIsNone(memFromUri.filename) 

186 self.assertEqual(memFromUri.origin, 0) 

187 self.assertTrue(memFromUri.isWriteable()) 

188 self.assertTrue(isEmptyDatabaseActuallyWriteable(memFromUri)) 

189 # We don't support SQLite URIs inside SQLAlchemy URIs. 

190 with self.assertRaises(NotImplementedError): 

191 SqliteDatabase.makeEngine(uri="sqlite:///:memory:?uri=true") 

192 # We don't support read-only in-memory databases. 

193 with self.assertRaises(NotImplementedError): 

194 SqliteDatabase.makeEngine(filename=None, writeable=False) 

195 

196 def testTransactionLocking(self): 

197 # This (inherited) test can't run on SQLite because of our use of an 

198 # aggressive locking strategy there. 

199 pass 

200 

201 

202class SqliteFileRegistryTests(RegistryTests): 

203 """Tests for `Registry` backed by a SQLite file-based database. 

204 

205 Notes 

206 ----- 

207 This is not a subclass of `unittest.TestCase` but to avoid repetition it 

208 defines methods that override `unittest.TestCase` methods. To make this 

209 work sublasses have to have this class first in the bases list. 

210 """ 

211 

212 def setUp(self): 

213 self.root = makeTestTempDir(TESTDIR) 

214 

215 def tearDown(self): 

216 removeTestTempDir(self.root) 

217 

218 @classmethod 

219 def getDataDir(cls) -> str: 

220 return os.path.normpath(os.path.join(os.path.dirname(__file__), "data", "registry")) 

221 

222 def make_butler(self, registry_config: RegistryConfig | None = None) -> Butler: 

223 config = Config() 

224 if registry_config is None: 

225 registry_config = self.makeRegistryConfig() 

226 config["registry"] = registry_config 

227 butler = makeTestRepo(self.root, config=config) 

228 cast(unittest.TestCase, self).enterContext(butler) 

229 return butler 

230 

231 

232class SqliteFileRegistryNameKeyCollMgrUUIDTestCase(SqliteFileRegistryTests, unittest.TestCase): 

233 """Tests for `Registry` backed by a SQLite file-based database. 

234 

235 This test case uses NameKeyCollectionManager and 

236 ByDimensionsDatasetRecordStorageManagerUUID. 

237 """ 

238 

239 collectionsManager = "lsst.daf.butler.registry.collections.nameKey.NameKeyCollectionManager" 

240 datasetsManager = ( 

241 "lsst.daf.butler.registry.datasets.byDimensions.ByDimensionsDatasetRecordStorageManagerUUID" 

242 ) 

243 

244 

245class ClonedSqliteFileRegistryNameKeyCollMgrUUIDTestCase( 

246 SqliteFileRegistryNameKeyCollMgrUUIDTestCase, unittest.TestCase 

247): 

248 """Test that NameKeyCollectionManager still works after cloning.""" 

249 

250 def make_butler(self, registry_config: RegistryConfig | None = None) -> Butler: 

251 original = super().make_butler(registry_config) 

252 return original.clone() 

253 

254 

255class SqliteFileRegistrySynthIntKeyCollMgrUUIDTestCase(SqliteFileRegistryTests, unittest.TestCase): 

256 """Tests for `Registry` backed by a SQLite file-based database. 

257 

258 This test case uses SynthIntKeyCollectionManager and 

259 ByDimensionsDatasetRecordStorageManagerUUID. 

260 """ 

261 

262 collectionsManager = "lsst.daf.butler.registry.collections.synthIntKey.SynthIntKeyCollectionManager" 

263 datasetsManager = ( 

264 "lsst.daf.butler.registry.datasets.byDimensions.ByDimensionsDatasetRecordStorageManagerUUID" 

265 ) 

266 

267 

268class SqliteMemoryRegistryTests(RegistryTests): 

269 """Tests for `Registry` backed by a SQLite in-memory database.""" 

270 

271 @classmethod 

272 def getDataDir(cls) -> str: 

273 return os.path.normpath(os.path.join(os.path.dirname(__file__), "data", "registry")) 

274 

275 def make_butler(self, registry_config: RegistryConfig | None = None) -> Butler: 

276 # This helper function always return in-memory registry 

277 # with default managers. 

278 if registry_config is None: 

279 registry_config = self.makeRegistryConfig() 

280 butler = create_populated_sqlite_registry(registry_config=registry_config) 

281 cast(unittest.TestCase, self).enterContext(butler) 

282 return butler 

283 

284 def testMissingAttributes(self): 

285 """Test for instantiating a registry against outdated schema which 

286 misses butler_attributes table. 

287 """ 

288 # TODO: Once we have stable gen3 schema everywhere this test can be 

289 # dropped (DM-27373). 

290 config = self.makeRegistryConfig() 

291 config["db"] = "sqlite://" 

292 with self.assertRaises(LookupError): 

293 _RegistryFactory(config).from_config() 

294 

295 

296class SqliteMemoryRegistryNameKeyCollMgrUUIDTestCase(unittest.TestCase, SqliteMemoryRegistryTests): 

297 """Tests for `Registry` backed by a SQLite in-memory database. 

298 

299 This test case uses NameKeyCollectionManager and 

300 ByDimensionsDatasetRecordStorageManagerUUID. 

301 """ 

302 

303 collectionsManager = "lsst.daf.butler.registry.collections.nameKey.NameKeyCollectionManager" 

304 datasetsManager = ( 

305 "lsst.daf.butler.registry.datasets.byDimensions.ByDimensionsDatasetRecordStorageManagerUUID" 

306 ) 

307 

308 

309class SqliteMemoryRegistrySynthIntKeyCollMgrUUIDTestCase(unittest.TestCase, SqliteMemoryRegistryTests): 

310 """Tests for `Registry` backed by a SQLite in-memory database. 

311 

312 This test case uses SynthIntKeyCollectionManager and 

313 ByDimensionsDatasetRecordStorageManagerUUID. 

314 """ 

315 

316 collectionsManager = "lsst.daf.butler.registry.collections.synthIntKey.SynthIntKeyCollectionManager" 

317 datasetsManager = ( 

318 "lsst.daf.butler.registry.datasets.byDimensions.ByDimensionsDatasetRecordStorageManagerUUID" 

319 ) 

320 

321 

322class SqliteMemoryRegistryAstropyIngestDateTestCase(unittest.TestCase, SqliteMemoryRegistryTests): 

323 """Tests for `Registry` backed by a SQLite in-memory database. 

324 

325 This test case uses version schema with ingest_date as nanoseconds instead 

326 of DATETIME. This tests can be removed when/if we switch to nanoseconds as 

327 default. 

328 """ 

329 

330 collectionsManager = "lsst.daf.butler.registry.collections.synthIntKey.SynthIntKeyCollectionManager" 

331 datasetsManager = { 

332 "cls": "lsst.daf.butler.registry.datasets.byDimensions.ByDimensionsDatasetRecordStorageManagerUUID", 

333 "schema_version": "2.0.0", 

334 } 

335 

336 

337if __name__ == "__main__": 

338 unittest.main()