Coverage for tests/test_sqlite.py: 41%

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 

34 

35import sqlalchemy 

36 

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

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

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

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

41from lsst.daf.butler.tests import makeTestRepo 

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

43 

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

45 

46 

47@contextmanager 

48def removeWritePermission(filename): 

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

50 mode = os.stat(filename).st_mode 

51 try: 

52 os.chmod(filename, stat.S_IREAD) 

53 yield 

54 finally: 

55 os.chmod(filename, mode) 

56 

57 

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

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

60 

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

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

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

64 """ 

65 try: 

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

67 table = context.addTable( 

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

69 ) 

70 # Drop created table so that schema remains empty. 

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

72 return True 

73 except Exception: 

74 return False 

75 

76 

77class SqliteFileDatabaseTestCase(unittest.TestCase, DatabaseTests): 

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

79 

80 def setUp(self): 

81 self.root = makeTestTempDir(TESTDIR) 

82 

83 def tearDown(self): 

84 removeTestTempDir(self.root) 

85 

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

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

88 engine = SqliteDatabase.makeEngine(filename=filename) 

89 return SqliteDatabase.fromEngine(engine=engine, origin=origin) 

90 

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

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

93 return SqliteDatabase.fromEngine(origin=database.origin, engine=engine, writeable=writeable) 

94 

95 @contextmanager 

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

97 with removeWritePermission(database.filename): 

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

99 

100 def testConnection(self): 

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

102 are equivalent. 

103 """ 

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

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

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

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

108 self.assertEqual(rwFromFilename.origin, 0) 

109 self.assertTrue(rwFromFilename.isWriteable()) 

110 self.assertTrue(isEmptyDatabaseActuallyWriteable(rwFromFilename)) 

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

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

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

114 self.assertEqual(rwFromUri.origin, 0) 

115 self.assertTrue(rwFromUri.isWriteable()) 

116 self.assertTrue(isEmptyDatabaseActuallyWriteable(rwFromUri)) 

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

118 with self.assertRaises(NotImplementedError): 

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

120 

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

122 with removeWritePermission(filename): 

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

124 roFromFilename = SqliteDatabase.fromEngine( 

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

126 ) 

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

128 self.assertEqual(roFromFilename.origin, 0) 

129 self.assertFalse(roFromFilename.isWriteable()) 

130 self.assertFalse(isEmptyDatabaseActuallyWriteable(roFromFilename)) 

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

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

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

134 self.assertEqual(roFromUri.origin, 0) 

135 self.assertFalse(roFromUri.isWriteable()) 

136 self.assertFalse(isEmptyDatabaseActuallyWriteable(roFromUri)) 

137 

138 def testTransactionLocking(self): 

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

140 # aggressive locking strategy there. 

141 pass 

142 

143 

144class SqliteMemoryDatabaseTestCase(unittest.TestCase, DatabaseTests): 

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

146 

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

148 engine = SqliteDatabase.makeEngine(filename=None) 

149 return SqliteDatabase.fromEngine(engine=engine, origin=origin) 

150 

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

152 return SqliteDatabase.fromEngine(origin=database.origin, engine=database._engine, writeable=writeable) 

153 

154 @contextmanager 

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

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

157 

158 def testConnection(self): 

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

160 are equivalent. 

161 """ 

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

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

164 self.assertIsNone(memFromFilename.filename) 

165 self.assertEqual(memFromFilename.origin, 0) 

166 self.assertTrue(memFromFilename.isWriteable()) 

167 self.assertTrue(isEmptyDatabaseActuallyWriteable(memFromFilename)) 

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

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

170 self.assertIsNone(memFromUri.filename) 

171 self.assertEqual(memFromUri.origin, 0) 

172 self.assertTrue(memFromUri.isWriteable()) 

173 self.assertTrue(isEmptyDatabaseActuallyWriteable(memFromUri)) 

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

175 with self.assertRaises(NotImplementedError): 

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

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

178 with self.assertRaises(NotImplementedError): 

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

180 

181 def testTransactionLocking(self): 

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

183 # aggressive locking strategy there. 

184 pass 

185 

186 

187class SqliteFileRegistryTests(RegistryTests): 

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

189 

190 Notes 

191 ----- 

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

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

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

195 """ 

196 

197 def setUp(self): 

198 self.root = makeTestTempDir(TESTDIR) 

199 

200 def tearDown(self): 

201 removeTestTempDir(self.root) 

202 

203 @classmethod 

204 def getDataDir(cls) -> str: 

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

206 

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

208 config = Config() 

209 if registry_config is None: 

210 registry_config = self.makeRegistryConfig() 

211 config["registry"] = registry_config 

212 return makeTestRepo(self.root, config=config) 

213 

214 

215class SqliteFileRegistryNameKeyCollMgrUUIDTestCase(SqliteFileRegistryTests, unittest.TestCase): 

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

217 

218 This test case uses NameKeyCollectionManager and 

219 ByDimensionsDatasetRecordStorageManagerUUID. 

220 """ 

221 

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

223 datasetsManager = ( 

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

225 ) 

226 

227 

228class ClonedSqliteFileRegistryNameKeyCollMgrUUIDTestCase( 

229 SqliteFileRegistryNameKeyCollMgrUUIDTestCase, unittest.TestCase 

230): 

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

232 

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

234 original = super().make_butler(registry_config) 

235 return original.clone() 

236 

237 

238class SqliteFileRegistrySynthIntKeyCollMgrUUIDTestCase(SqliteFileRegistryTests, unittest.TestCase): 

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

240 

241 This test case uses SynthIntKeyCollectionManager and 

242 ByDimensionsDatasetRecordStorageManagerUUID. 

243 """ 

244 

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

246 datasetsManager = ( 

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

248 ) 

249 

250 

251class SqliteMemoryRegistryTests(RegistryTests): 

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

253 

254 @classmethod 

255 def getDataDir(cls) -> str: 

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

257 

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

259 # This helper function always return in-memory registry 

260 # with default managers. 

261 if registry_config is None: 

262 registry_config = self.makeRegistryConfig() 

263 return create_populated_sqlite_registry(registry_config=registry_config) 

264 

265 def testMissingAttributes(self): 

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

267 misses butler_attributes table. 

268 """ 

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

270 # dropped (DM-27373). 

271 config = self.makeRegistryConfig() 

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

273 with self.assertRaises(LookupError): 

274 _RegistryFactory(config).from_config() 

275 

276 

277class SqliteMemoryRegistryNameKeyCollMgrUUIDTestCase(unittest.TestCase, SqliteMemoryRegistryTests): 

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

279 

280 This test case uses NameKeyCollectionManager and 

281 ByDimensionsDatasetRecordStorageManagerUUID. 

282 """ 

283 

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

285 datasetsManager = ( 

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

287 ) 

288 

289 

290class SqliteMemoryRegistrySynthIntKeyCollMgrUUIDTestCase(unittest.TestCase, SqliteMemoryRegistryTests): 

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

292 

293 This test case uses SynthIntKeyCollectionManager and 

294 ByDimensionsDatasetRecordStorageManagerUUID. 

295 """ 

296 

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

298 datasetsManager = ( 

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

300 ) 

301 

302 

303class SqliteMemoryRegistryAstropyIngestDateTestCase(unittest.TestCase, SqliteMemoryRegistryTests): 

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

305 

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

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

308 default. 

309 """ 

310 

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

312 datasetsManager = { 

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

314 "schema_version": "2.0.0", 

315 } 

316 

317 

318if __name__ == "__main__": 

319 unittest.main()