Coverage for tests / test_database_context.py: 31%

1# This file is part of felis. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://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 <https://www.gnu.org/licenses/>. 

21 

22import os 

23import unittest 

24 

25import sqlalchemy 

26 

27try: 

28 from testing.postgresql import Postgresql # type: ignore 

29except ImportError: 

30 Postgresql = None 

31 

32from felis.datamodel import Schema 

33from felis.db.database_context import ( 

34 DatabaseContext, 

35 DatabaseContextError, 

36 DatabaseContextFactory, 

37 MockContext, 

38 MySQLContext, 

39 PostgreSQLContext, 

40 create_database_context, 

41) 

42from felis.metadata import MetaDataBuilder 

43 

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

45TEST_YAML = os.path.join(TEST_DIR, "data", "sales.yaml") 

46 

47 

48class BaseDatabaseContextTest: 

49 """Base tests of database context.""" 

50 

51 def setUp(self) -> None: 

52 """Set up the test case.""" 

53 self._schema = Schema.from_uri(TEST_YAML) 

54 self._metadata = MetaDataBuilder(self._schema).build() 

55 self._engine_url: str 

56 self._dialect_name: str 

57 

58 def _create_database_context(self) -> DatabaseContext: 

59 return create_database_context(self._engine_url, self._metadata) 

60 

61 def test_database_context(self): 

62 """Test database context with SQLite.""" 

63 with self._create_database_context() as db_ctx: 

64 self.assertIsNotNone(db_ctx) 

65 self.assertIsNotNone(db_ctx.metadata) 

66 if not isinstance(db_ctx, MockContext): 

67 # Only a non-mock connection has a valid engine object. 

68 self.assertIsNotNone(db_ctx.engine) 

69 self.assertEqual(db_ctx.engine.dialect.name, self._dialect_name) 

70 

71 # Drop first in case it exists from a previous test 

72 db_ctx.drop() 

73 

74 # Initialize the database 

75 db_ctx.initialize() 

76 

77 # Create all tables 

78 db_ctx.create_all() 

79 

80 # Drop indexes 

81 db_ctx.drop_indexes() 

82 

83 # Create indexes 

84 db_ctx.create_indexes() 

85 

86 # Determine schema prefix (none if schema name is not set) 

87 if db_ctx.metadata.schema: 

88 schema_prefix = f"{db_ctx.metadata.schema}." 

89 else: 

90 schema_prefix = "" 

91 

92 # Execute a simple query from a string 

93 db_ctx.execute(f"SELECT * FROM {schema_prefix}customers") 

94 

95 # Execute a simple query from a TextClause 

96 db_ctx.execute(sqlalchemy.text(f"SELECT * FROM {schema_prefix}orders")) 

97 

98 # Execute a query that raises an error, except for MockContext 

99 if not isinstance(db_ctx, MockContext): 

100 with self.assertRaises(DatabaseContextError): 

101 db_ctx.execute("SELECT * FROM non_existent_table") 

102 

103 # Drop all tables 

104 db_ctx.drop() 

105 

106 def test_initialize_is_idempotent(self): 

107 """Test that attempting to create a schema that already exists does not 

108 raise an error. 

109 """ 

110 with self._create_database_context() as db_ctx: 

111 # Ensure database does not exist by first dropping 

112 db_ctx.drop() 

113 

114 # Initialize the database 

115 db_ctx.initialize() 

116 

117 # Second initialization should not raise an error. 

118 db_ctx.initialize() 

119 

120 # Clean up 

121 db_ctx.drop() 

122 

123 def test_drop_is_idempotent(self): 

124 """Test that dropping a schema that does not exist does not raise an 

125 error. 

126 """ 

127 with self._create_database_context() as db_ctx: 

128 # Initialize the database 

129 db_ctx.initialize() 

130 

131 # Drop the schema 

132 db_ctx.drop() 

133 

134 # Second drop should not raise an error. 

135 db_ctx.drop() 

136 

137 

138class SQLiteTestCase(BaseDatabaseContextTest, unittest.TestCase): 

139 """Tests of database context using SQLite dialect.""" 

140 

141 def setUp(self) -> None: 

142 """Set up the test case.""" 

143 super().setUp() 

144 self._dialect_name = "sqlite" 

145 self._engine_url = "sqlite:///:memory:" 

146 

147 

148@unittest.skipIf(Postgresql is None, "testing.postgresql is not installed") 

149class PostgreSQLTestCase(BaseDatabaseContextTest, unittest.TestCase): 

150 """Tests of database context using PostgreSQL dialect.""" 

151 

152 def setUp(self) -> None: 

153 """Set up the test case.""" 

154 super().setUp() 

155 self._dialect_name = "postgresql" 

156 self._postgres = Postgresql() 

157 self._engine_url = self._postgres.url() 

158 

159 def test_missing_schema_name(self): 

160 """Test that a missing schema name raises an error when using 

161 a Postgres context. 

162 """ 

163 metadata = MetaDataBuilder(self._schema, apply_schema_to_metadata=False).build() 

164 with self.assertRaises(DatabaseContextError): 

165 PostgreSQLContext(sqlalchemy.engine.make_url(self._engine_url), metadata) 

166 

167 

168class MySQLTestCase(BaseDatabaseContextTest, unittest.TestCase): 

169 """Tests of MySQL database context.""" 

170 

171 # Environment variable name for MySQL engine URL 

172 _env_name = "_FELIS_MYSQL_ENGINE_URL" 

173 

174 def setUp(self) -> None: 

175 super().setUp() 

176 try: 

177 mysql_engine_url = os.environ[f"{self._env_name}"] 

178 except KeyError: 

179 raise unittest.SkipTest(f"{self._env_name} is not set in the environment; skipping MySQL tests.") 

180 if mysql_engine_url is None or mysql_engine_url == "": 

181 raise ValueError(f"Value of {self._env_name} from environment is invalid") 

182 self._dialect_name = "mysql" 

183 self._engine_url = mysql_engine_url 

184 

185 def test_missing_schema_name(self): 

186 """Test that a missing schema name raises an error when using 

187 a MySQL context. 

188 """ 

189 metadata = MetaDataBuilder(self._schema, apply_schema_to_metadata=False).build() 

190 with self.assertRaises(DatabaseContextError): 

191 MySQLContext(sqlalchemy.engine.make_url(self._engine_url), metadata) 

192 

193 

194class MockTestCase(BaseDatabaseContextTest, unittest.TestCase): 

195 """Tests of mock database context.""" 

196 

197 def setUp(self) -> None: 

198 super().setUp() 

199 # This URL should result in a mock connection being setup. 

200 self._engine_url = "sqlite://" 

201 

202 def test_mock_connection_engine_error(self): 

203 """Test that attempting to access the engine of a mock context throws 

204 an error. 

205 """ 

206 db_ctx = create_database_context(self._engine_url, self._metadata) 

207 with self.assertRaises(DatabaseContextError): 

208 _ = db_ctx.engine 

209 

210 

211class DatabaseContextTestCase(unittest.TestCase): 

212 """Test that a mismatch between the engine and database context correctly 

213 throws an error. 

214 """ 

215 

216 def test_create_with_bad_url(self): 

217 """Test that using a SQLite engine with a Postgres context correctly 

218 throws an error. 

219 """ 

220 with self.assertRaises(DatabaseContextError): 

221 engine = sqlalchemy.engine.make_url("sqlite:///:memory:") 

222 PostgreSQLContext(engine, sqlalchemy.MetaData(schema="test_schema")) 

223 

224 def test_create_with_bad_dialect(self): 

225 """Test that attempting to create a database context with an 

226 unsupported dialect raises an error. 

227 """ 

228 engine_url = "oracle+cx_oracle://user:password@host:1521/service_name" 

229 metadata = sqlalchemy.MetaData() 

230 with self.assertRaises(DatabaseContextError): 

231 create_database_context(engine_url, metadata) 

232 

233 

234class SupportedDialectsTestCase(unittest.TestCase): 

235 """Test supported dialects.""" 

236 

237 def test_supported_dialects(self) -> None: 

238 """Test that supported dialects are correctly reported.""" 

239 supported_dialects = DatabaseContextFactory.get_supported_dialects() 

240 self.assertIn("sqlite", supported_dialects) 

241 self.assertIn("postgresql", supported_dialects) 

242 self.assertIn("mysql", supported_dialects) 

243 

244 def test_supported_dialect_with_driver(self) -> None: 

245 """Test that supported dialects with drivers are correctly 

246 normalized. 

247 """ 

248 supported_dialects = DatabaseContextFactory.get_supported_dialects() 

249 # Test that postgresql+psycopg2 normalizes to postgresql 

250 self.assertIn("postgresql", supported_dialects) 

251 # Test that mysql+mysqlconnector normalizes to mysql 

252 self.assertIn("mysql", supported_dialects) 

253 

254 def test_unsupported_dialect(self) -> None: 

255 """Test that an unsupported dialect raises an error.""" 

256 metadata = sqlalchemy.MetaData() 

257 with self.assertRaises(DatabaseContextError): 

258 create_database_context("oracle+cx_oracle://user:pass@host/db", metadata) 

259 with self.assertRaises(DatabaseContextError): 

260 create_database_context("oracle://user:pass@host/db", metadata) 

261 

262 

263if __name__ == "__main__": 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true

264 unittest.main()