Coverage for tests/test_tap_schema_postgres.py: 29%

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 gc 

23import os 

24import unittest 

25 

26from sqlalchemy.engine import create_engine 

27from sqlalchemy.exc import SQLAlchemyError 

28from sqlalchemy.schema import CreateSchema 

29 

30from felis.datamodel import Schema 

31from felis.db.utils import DatabaseContext 

32from felis.metadata import MetaDataBuilder 

33from felis.tap_schema import DataLoader, TableManager 

34 

35try: 

36 from testing.postgresql import Postgresql 

37except ImportError: 

38 Postgresql = None 

39 

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

41TEST_SALES = os.path.join(TEST_DIR, "data", "sales.yaml") 

42TEST_TAP_SCHEMA_NONSTD = os.path.join(TEST_DIR, "data", "test_tap_schema_nonstandard.yaml") 

43 

44 

45class TestTapSchemaPostgresql(unittest.TestCase): 

46 """Test TAP_SCHEMA for PostgreSQL""" 

47 

48 def setUp(self) -> None: 

49 """Set up a local PostgreSQL database and a test schema.""" 

50 # Skip the test if the testing.postgresql package is not installed. 

51 if not Postgresql: 

52 self.skipTest("testing.postgresql not installed") 

53 

54 # Start a PostgreSQL database for testing. 

55 self.postgresql = Postgresql() 

56 url = self.postgresql.url() 

57 self.engine = create_engine(url) 

58 

59 # Setup a test schema. 

60 self.test_schema = Schema.from_uri(TEST_SALES) 

61 

62 def test_create_metadata(self) -> None: 

63 """Test loading of data into a PostgreSQL TAP_SCHEMA database created 

64 by the `~felis.tap_schema.TableManager`. 

65 """ 

66 try: 

67 # Create the TAP_SCHEMA database. 

68 mgr = TableManager() 

69 mgr.initialize_database(self.engine) 

70 

71 # Load the test data into the database. 

72 loader = DataLoader(self.test_schema, mgr, self.engine, 1) 

73 loader.load() 

74 finally: 

75 # Drop the schema. 

76 DatabaseContext(metadata=mgr.metadata, engine=self.engine).drop() 

77 

78 def test_reflect_database(self) -> None: 

79 """Test reflecting an existing PostgreSQL TAP_SCHEMA database into a 

80 `~felis.tap_schema.TableManager`. 

81 """ 

82 try: 

83 # Build the TAP_SCHEMA database independently of the TableManager. 

84 schema = TableManager.load_schema_resource() 

85 md = MetaDataBuilder(schema).build() 

86 with self.engine.connect() as conn: 

87 trans = conn.begin() 

88 try: 

89 print(f"Creating schema '{schema.name}'") 

90 conn.execute(CreateSchema(schema.name, if_not_exists=False)) 

91 trans.commit() 

92 except SQLAlchemyError as e: 

93 trans.rollback() 

94 self.fail(f"Failed to create schema: {e}") 

95 try: 

96 print(f"Creating tables in schema: {md.schema}") 

97 md.create_all(self.engine) 

98 except SQLAlchemyError as e: 

99 self.fail(f"Failed to create database: {e}") 

100 

101 # Reflect the existing database into a TableManager. 

102 mgr = TableManager(engine=self.engine) 

103 self.assertIsNotNone(mgr.metadata) 

104 self.assertGreater(len(mgr.metadata.tables), 0) 

105 table_names = set( 

106 [table_name.replace(f"{schema.name}.", "") for table_name in mgr.metadata.tables.keys()] 

107 ) 

108 self.assertEqual(table_names, set(TableManager.get_table_names_std())) 

109 

110 # See if test data can be loaded successfully using the existing 

111 # database. 

112 loader = DataLoader(self.test_schema, mgr, self.engine, 1) 

113 loader.load() 

114 finally: 

115 # Drop the schema. 

116 DatabaseContext(metadata=mgr.metadata, engine=self.engine).drop() 

117 

118 def test_nonstandard_names(self) -> None: 

119 """Test the TAP table manager class with non-standard names for the 

120 schema and columns, which are present in the test YAML file used 

121 to create the TAP_SCHEMA database. 

122 """ 

123 try: 

124 with open(TEST_TAP_SCHEMA_NONSTD) as file: 

125 sch = Schema.from_stream(file, context={"id_generation": True}) 

126 md = MetaDataBuilder(sch).build() 

127 ctx = DatabaseContext(md, self.engine) 

128 ctx.initialize() 

129 ctx.create_all() 

130 

131 postfix = "11" 

132 mgr = TableManager(engine=self.engine, table_name_postfix=postfix, schema_name=sch.name) 

133 for table_name in mgr.get_table_names_std(): 

134 table = mgr[table_name] 

135 self.assertEqual(table.name, f"{table_name}{postfix}".replace(f"{sch.name}", "")) 

136 finally: 

137 if ctx: 

138 ctx.drop() 

139 

140 def test_bad_engine(self) -> None: 

141 """Test the TableManager class with an invalid engine.""" 

142 bad_engine = create_engine("postgresql+psycopg2://fake_user:fake_password@fake_host:5555") 

143 with self.assertRaises(SQLAlchemyError): 

144 TableManager(engine=bad_engine) 

145 

146 def tearDown(self) -> None: 

147 """Tear down the test case.""" 

148 gc.collect() 

149 self.engine.dispose() 

150 

151 

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

153 unittest.main()