Coverage for tests/test_tap_schema.py: 18%
392 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-24 01:22 -0800
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/>.
22import os
23import shutil
24import tempfile
25import unittest
26from typing import Any
28from sqlalchemy import create_engine, select
30from felis.datamodel import Schema
31from felis.tap_schema import DataLoader, TableManager
33TEST_DIR = os.path.dirname(__file__)
34TEST_SALES = os.path.join(TEST_DIR, "data", "sales.yaml")
35TEST_TAP_SCHEMA = os.path.join(TEST_DIR, "data", "test_tap_schema.yaml")
36TEST_COMPOSITE_KEYS = os.path.join(TEST_DIR, "data", "test_composite_keys.yaml")
39class TableManagerTestCase(unittest.TestCase):
40 """Test the `TableManager` class."""
42 def setUp(self) -> None:
43 """Set up the test case."""
44 with open(TEST_SALES) as test_file:
45 self.schema = Schema.from_stream(test_file)
47 def test_create_table_manager(self) -> None:
48 """Test the TAP table manager class."""
49 mgr = TableManager()
51 schema_name = mgr.schema_name
53 # Check the created metadata and tables.
54 self.assertNotEqual(len(mgr.metadata.tables), 0)
55 self.assertEqual(mgr.metadata.schema, schema_name)
56 for table_name in mgr.get_table_names_std():
57 mgr[table_name]
59 # Make sure that creating a new table manager works when one has
60 # already been created.
61 mgr = TableManager()
63 def test_table_name_postfix(self) -> None:
64 """Test the table name postfix."""
65 mgr = TableManager(apply_schema_to_metadata=False, table_name_postfix="_test")
66 for table_name in mgr.metadata.tables:
67 self.assertTrue(table_name.endswith("_test"))
70class DataLoaderTestCase(unittest.TestCase):
71 """Test the `DataLoader` class."""
73 def setUp(self) -> None:
74 """Set up the test case."""
75 with open(TEST_TAP_SCHEMA) as test_file:
76 self.schema = Schema.from_stream(test_file, context={"id_generation": True})
78 self.tmpdir = tempfile.mkdtemp(dir=TEST_DIR)
80 def tearDown(self) -> None:
81 """Clean up temporary directory."""
82 shutil.rmtree(self.tmpdir, ignore_errors=True)
84 def test_sqlite(self) -> None:
85 """Test the `DataLoader` using an in-memory SQLite database."""
86 engine = create_engine("sqlite:///:memory:")
88 mgr = TableManager(apply_schema_to_metadata=False)
89 mgr.initialize_database(engine)
91 loader = DataLoader(self.schema, mgr, engine)
92 loader.load()
94 def test_sql_output(self) -> None:
95 """Test printing SQL to stdout and writing SQL to a file."""
96 engine = create_engine("sqlite:///:memory:")
97 mgr = TableManager(apply_schema_to_metadata=False)
98 loader = DataLoader(self.schema, mgr, engine, dry_run=True, print_sql=True)
99 loader.load()
101 sql_path = os.path.join(self.tmpdir, "test_tap_schema_print_sql.sql")
102 loader = DataLoader(self.schema, mgr, engine, dry_run=True, print_sql=True, output_path=sql_path)
103 loader.load()
105 self.assertTrue(os.path.exists(sql_path))
106 with open(sql_path) as sql_file:
107 sql_data = sql_file.read()
108 insert_count = sql_data.count("INSERT INTO")
109 self.assertEqual(
110 insert_count,
111 22,
112 f"Expected 22 'INSERT INTO' statements, found {insert_count}",
113 )
115 def test_unique_keys(self) -> None:
116 """Test generation of unique foreign keys."""
117 engine = create_engine("sqlite:///:memory:")
119 mgr = TableManager(apply_schema_to_metadata=False)
120 mgr.initialize_database(engine)
122 loader = DataLoader(self.schema, mgr, engine, unique_keys=True)
123 loader.load()
125 keys_data = mgr.select(engine, "keys")
126 self.assertGreaterEqual(len(keys_data), 1)
127 for row in keys_data:
128 self.assertTrue(row["key_id"].startswith(f"{self.schema.name}_"))
130 key_columns_data = mgr.select(engine, "key_columns")
131 self.assertGreaterEqual(len(key_columns_data), 1)
132 for row in key_columns_data:
133 self.assertTrue(row["key_id"].startswith(f"{self.schema.name}_"))
135 def test_select_with_filter(self) -> None:
136 """Test selecting rows with a filter."""
137 engine = create_engine("sqlite:///:memory:")
138 mgr = TableManager(apply_schema_to_metadata=False)
139 mgr.initialize_database(engine)
140 loader = DataLoader(self.schema, mgr, engine, unique_keys=True)
141 loader.load()
143 rows = mgr.select(engine, "columns", "table_name = 'test_schema.table1'")
144 self.assertEqual(len(rows), 16)
147def _find_row(rows: list[dict[str, Any]], column_name: str, value: str) -> dict[str, Any]:
148 next_row = next(
149 (row for row in rows if row[column_name] == value),
150 None,
151 )
152 assert next_row is not None
153 assert isinstance(next_row, dict)
154 return next_row
157class TapSchemaSqliteSetup:
158 """Set up the TAP_SCHEMA SQLite database for testing.
160 Parameters
161 ----------
162 test_file_path:
163 Path to the TAP_SCHEMA test file.
165 context
166 Context for the schema. Default is an empty dictionary.
167 """
169 def __init__(self, test_file_path: str, context: dict = {}) -> None:
170 with open(test_file_path) as test_file:
171 self._schema = Schema.from_stream(test_file, context=context)
173 self._engine = create_engine("sqlite:///:memory:")
175 mgr = TableManager(apply_schema_to_metadata=False)
176 mgr.initialize_database(self._engine)
177 self._mgr = mgr
179 loader = DataLoader(self._schema, mgr, self._engine, tap_schema_index=2)
180 loader.load()
182 @property
183 def schema(self) -> Schema:
184 """Return the schema."""
185 return self._schema
187 @property
188 def engine(self) -> Any:
189 """Return the engine."""
190 return self._engine
192 @property
193 def mgr(self) -> TableManager:
194 """Return the table manager."""
195 return self._mgr
198class TapSchemaDataTest(unittest.TestCase):
199 """Test the validity of generated TAP SCHEMA data."""
201 def setUp(self) -> None:
202 """Set up the test case."""
203 self.tap_schema_setup = TapSchemaSqliteSetup(TEST_TAP_SCHEMA, context={"id_generation": True})
205 def test_schemas(self) -> None:
206 schemas_table = self.tap_schema_setup.mgr["schemas"]
207 with self.tap_schema_setup.engine.connect() as connection:
208 result = connection.execute(select(schemas_table))
209 schema_data = [row._asdict() for row in result]
211 self.assertEqual(len(schema_data), 1)
213 schema = schema_data[0]
214 self.assertEqual(schema["schema_name"], "test_schema")
215 self.assertEqual(schema["description"], "Test schema")
216 self.assertEqual(schema["utype"], "Schema")
217 self.assertEqual(schema["schema_index"], 2)
219 def test_tables(self) -> None:
220 tables_table = self.tap_schema_setup.mgr["tables"]
221 with self.tap_schema_setup.engine.connect() as connection:
222 result = connection.execute(select(tables_table))
223 table_data = [row._asdict() for row in result]
225 self.assertEqual(len(table_data), 2)
227 table = table_data[0]
228 assert isinstance(table, dict)
229 self.assertEqual(table["schema_name"], "test_schema")
230 self.assertEqual(table["table_name"], f"{self.tap_schema_setup.schema.name}.table1")
231 self.assertEqual(table["table_type"], "table")
232 self.assertEqual(table["utype"], "Table")
233 self.assertEqual(table["description"], "Test table 1")
234 self.assertEqual(table["table_index"], 2)
236 def test_columns(self) -> None:
237 columns_table = self.tap_schema_setup.mgr["columns"]
238 with self.tap_schema_setup.engine.connect() as connection:
239 result = connection.execute(select(columns_table))
240 column_data = [row._asdict() for row in result]
242 table1_rows = [
243 row for row in column_data if row["table_name"] == f"{self.tap_schema_setup.schema.name}.table1"
244 ]
245 self.assertNotEqual(len(table1_rows), 0)
247 boolean_col = _find_row(table1_rows, "column_name", "boolean_field")
248 self.assertEqual(boolean_col["datatype"], "boolean")
249 self.assertEqual(boolean_col["arraysize"], None)
251 byte_col = _find_row(table1_rows, "column_name", "byte_field")
252 self.assertEqual(byte_col["datatype"], "unsignedByte")
253 self.assertEqual(byte_col["arraysize"], None)
255 short_col = _find_row(table1_rows, "column_name", "short_field")
256 self.assertEqual(short_col["datatype"], "short")
257 self.assertEqual(short_col["arraysize"], None)
259 int_col = _find_row(table1_rows, "column_name", "int_field")
260 self.assertEqual(int_col["datatype"], "int")
261 self.assertEqual(int_col["arraysize"], None)
263 float_col = _find_row(table1_rows, "column_name", "float_field")
264 self.assertEqual(float_col["datatype"], "float")
265 self.assertEqual(float_col["arraysize"], None)
267 double_col = _find_row(table1_rows, "column_name", "double_field")
268 self.assertEqual(double_col["datatype"], "double")
269 self.assertEqual(double_col["arraysize"], None)
271 long_col = _find_row(table1_rows, "column_name", "long_field")
272 self.assertEqual(long_col["datatype"], "long")
273 self.assertEqual(long_col["arraysize"], None)
275 unicode_col = _find_row(table1_rows, "column_name", "unicode_field")
276 self.assertEqual(unicode_col["datatype"], "unicodeChar")
277 self.assertEqual(unicode_col["arraysize"], "128*")
279 binary_col = _find_row(table1_rows, "column_name", "binary_field")
280 self.assertEqual(binary_col["datatype"], "unsignedByte")
281 self.assertEqual(binary_col["arraysize"], "1024*")
283 ts = _find_row(table1_rows, "column_name", "timestamp_field")
284 self.assertEqual(ts["datatype"], "char")
285 self.assertEqual(ts["xtype"], "timestamp")
286 self.assertEqual(ts["description"], "Timestamp field")
287 self.assertEqual(ts["utype"], "Obs:Timestamp")
288 self.assertEqual(ts["unit"], "s")
289 self.assertEqual(ts["ucd"], "time.epoch")
290 self.assertEqual(ts["principal"], 1)
291 self.assertEqual(ts["std"], 1)
292 self.assertEqual(ts["column_index"], 42)
293 self.assertEqual(ts["size"], None)
294 self.assertEqual(ts["arraysize"], "*")
296 char_col = _find_row(table1_rows, "column_name", "char_field")
297 self.assertEqual(char_col["datatype"], "char")
298 self.assertEqual(char_col["arraysize"], "64")
300 str_col = _find_row(table1_rows, "column_name", "string_field")
301 self.assertEqual(str_col["datatype"], "char")
302 self.assertEqual(str_col["arraysize"], "256*")
304 txt_col = _find_row(table1_rows, "column_name", "text_field")
305 self.assertEqual(txt_col["datatype"], "char")
306 self.assertEqual(txt_col["arraysize"], "*")
308 def test_keys(self) -> None:
309 keys_table = self.tap_schema_setup.mgr["keys"]
310 with self.tap_schema_setup.engine.connect() as connection:
311 result = connection.execute(select(keys_table))
312 key_data = [row._asdict() for row in result]
314 self.assertEqual(len(key_data), 1)
316 key = key_data[0]
317 assert isinstance(key, dict)
319 self.assertEqual(key["key_id"], "fk_table1_to_table2")
320 self.assertEqual(key["from_table"], f"{self.tap_schema_setup.schema.name}.table1")
321 self.assertEqual(key["target_table"], f"{self.tap_schema_setup.schema.name}.table2")
322 self.assertEqual(key["description"], "Foreign key from table1 to table2")
323 self.assertEqual(key["utype"], "ForeignKey")
325 def test_key_columns(self) -> None:
326 key_columns_table = self.tap_schema_setup.mgr["key_columns"]
327 with self.tap_schema_setup.engine.connect() as connection:
328 result = connection.execute(select(key_columns_table))
329 key_column_data = [row._asdict() for row in result]
331 self.assertEqual(len(key_column_data), 1)
333 key_column = key_column_data[0]
334 assert isinstance(key_column, dict)
336 self.assertEqual(key_column["key_id"], "fk_table1_to_table2")
337 self.assertEqual(key_column["from_column"], "fk")
338 self.assertEqual(key_column["target_column"], "id")
340 def test_bad_table_name(self) -> None:
341 """Test getting a bad TAP_SCHEMA table name."""
342 with self.assertRaises(KeyError):
343 self.tap_schema_setup.mgr["bad_table"]
346class ForceUnboundArraySizeTest(unittest.TestCase):
347 """Test that arraysize for appropriate types is set to '*' when the
348 ``force_unboundeded_arraysize`` context flag is set to ``True``.
349 """
351 def setUp(self) -> None:
352 """Set up the test case."""
353 self.tap_schema_setup = TapSchemaSqliteSetup(
354 TEST_TAP_SCHEMA, context={"id_generation": True, "force_unbounded_arraysize": True}
355 )
357 def test_force_unbounded_arraysize(self) -> None:
358 """Test that unbounded arraysize is set to None."""
359 columns_table = self.tap_schema_setup.mgr["columns"]
360 with self.tap_schema_setup.engine.connect() as connection:
361 result = connection.execute(select(columns_table))
362 column_data = [row._asdict() for row in result]
364 table1_rows = [
365 row for row in column_data if row["table_name"] == f"{self.tap_schema_setup.schema.name}.table1"
366 ]
367 for row in table1_rows:
368 if row["column_name"] in ["string_field", "text_field", "unicode_field", "binary_field"]:
369 self.assertEqual(row["arraysize"], "*")
372class CompositeKeysTestCase(unittest.TestCase):
373 """Test the handling of composite foreign keys."""
375 def setUp(self) -> None:
376 """Set up the test case."""
377 self.tap_schema_setup = TapSchemaSqliteSetup(TEST_COMPOSITE_KEYS, context={"id_generation": True})
379 # Fetch the keys and key_columns data from the TAP_SCHEMA tables.
380 keys_table = self.tap_schema_setup.mgr["keys"]
381 key_columns_table = self.tap_schema_setup.mgr["key_columns"]
382 with self.tap_schema_setup.engine.connect() as connection:
383 key_columns_result = connection.execute(select(key_columns_table))
384 self.key_columns_data = [row._asdict() for row in key_columns_result]
386 keys_result = connection.execute(select(keys_table))
387 self.keys_data = [row._asdict() for row in keys_result]
389 def test_keys(self) -> None:
390 """Test that composite keys are handled correctly by inspecting the
391 data in the generated TAP_SCHEMA ``keys`` table.
392 """
393 print(f"\nComposite keys data: {self.keys_data}")
395 self.assertEqual(len(self.keys_data), 1)
397 self.assertEqual(
398 self.keys_data[0],
399 {
400 "key_id": "fk_composite",
401 "from_table": "test_composite_keys.table1",
402 "target_table": "test_composite_keys.table2",
403 "utype": "ForeignKey",
404 "description": "Composite foreign key from table1 to table2",
405 },
406 )
408 def test_key_columns(self) -> None:
409 """Test that composite keys are handled correctly by inspecting the
410 data in the generated TAP_SCHEMA ``key_columns`` table.
411 """
412 print(f"\nComposite key columns data: {self.key_columns_data}")
414 self.assertEqual(len(self.key_columns_data), 2)
416 key_columns_row1 = self.key_columns_data[0]
417 self.assertEqual(
418 key_columns_row1, {"key_id": "fk_composite", "from_column": "id1", "target_column": "id1"}
419 )
421 key_columns_row2 = self.key_columns_data[1]
422 self.assertEqual(
423 key_columns_row2, {"key_id": "fk_composite", "from_column": "id2", "target_column": "id2"}
424 )
427class TableManagerExtensionsTestCase(unittest.TestCase):
428 """Test the `TableManager` class with extensions."""
430 def setUp(self) -> None:
431 """Set up the test case."""
432 self.tmpdir = tempfile.mkdtemp(dir=TEST_DIR)
434 self.extensions_path = os.path.join(self.tmpdir, "test_extensions.yaml")
435 extensions_content = """
436name: test_extensions
437description: Test TAP_SCHEMA extensions
439tables:
440 - name: schemas
441 description: Extensions to schemas table
442 columns:
443 - name: owner_id
444 datatype: char
445 length: 32
446 nullable: true
447 description: "Owner identifier"
448 - name: read_anon
449 datatype: int
450 nullable: true
451 description: "Anon read flag"
453 - name: tables
454 description: Extensions to tables table
455 columns:
456 - name: api_created
457 datatype: int
458 nullable: true
459 description: "API created flag"
460"""
461 with open(self.extensions_path, "w") as f:
462 f.write(extensions_content)
464 def tearDown(self) -> None:
465 """Clean up temporary directory."""
466 shutil.rmtree(self.tmpdir, ignore_errors=True)
468 def test_extensions_applied(self) -> None:
469 mgr = TableManager(extensions_path=self.extensions_path)
471 schemas_table = mgr["schemas"]
472 self.assertIn("owner_id", schemas_table.c)
473 self.assertIn("read_anon", schemas_table.c)
475 tables_table = mgr["tables"]
476 self.assertIn("api_created", tables_table.c)
478 def test_extensions_column_count(self) -> None:
479 mgr_without = TableManager()
480 mgr_with = TableManager(extensions_path=self.extensions_path)
482 schemas_before = len(mgr_without["schemas"].c)
483 schemas_after = len(mgr_with["schemas"].c)
484 self.assertEqual(schemas_after, schemas_before + 2)
486 tables_before = len(mgr_without["tables"].c)
487 tables_after = len(mgr_with["tables"].c)
488 self.assertEqual(tables_after, tables_before + 1)
490 def test_extensions_with_data_loader(self) -> None:
491 engine = create_engine("sqlite:///:memory:")
493 mgr = TableManager(apply_schema_to_metadata=False, extensions_path=self.extensions_path)
494 mgr.initialize_database(engine)
496 with open(TEST_TAP_SCHEMA) as test_file:
497 schema = Schema.from_stream(test_file, context={"id_generation": True})
499 loader = DataLoader(schema, mgr, engine)
500 loader.load()
502 schemas_table = mgr["schemas"]
503 with engine.connect() as connection:
504 result = connection.execute(select(schemas_table))
505 row = result.fetchone()
506 self.assertIn("owner_id", row._fields)
507 self.assertIn("read_anon", row._fields)
509 def test_invalid_extensions_file(self) -> None:
510 invalid_path = os.path.join(self.tmpdir, "nonexistent.yaml")
512 with self.assertRaises(ValueError):
513 TableManager(extensions_path=invalid_path)
515 def test_empty_extensions(self) -> None:
516 empty_extensions_path = os.path.join(self.tmpdir, "empty_extensions.yaml")
517 with open(empty_extensions_path, "w") as f:
518 f.write("name: empty_extensions\ntables: []\n")
520 mgr = TableManager(extensions_path=empty_extensions_path)
521 self.assertIsNotNone(mgr["schemas"])
523 def test_extensions_with_null_table_extensions(self) -> None:
524 null_extensions_path = os.path.join(self.tmpdir, "null_extensions.yaml")
525 with open(null_extensions_path, "w") as f:
526 f.write("""
527name: null_extensions
528tables:
529 - name: schemas
530 columns: []
531 - name: tables
532 columns: []
533 - name: columns
534 columns:
535 - name: test_col
536 datatype: int
537""")
539 mgr = TableManager(extensions_path=null_extensions_path)
541 columns_table = mgr["columns"]
542 self.assertIn("test_col", columns_table.c)
544 schemas_table = mgr["schemas"]
545 self.assertNotIn("owner_id", schemas_table.c)
547 def test_extensions_invalid_column_missing_name(self) -> None:
548 invalid_name_path = os.path.join(self.tmpdir, "invalid_name.yaml")
549 with open(invalid_name_path, "w") as f:
550 f.write("""
551 name: invalid_name
552 tables:
553 - name: schemas
554 columns:
555 - datatype: int
556 description: "Missing name"
557 - name: some_column
558 datatype: int
559 """)
560 with self.assertRaises(KeyError):
561 TableManager(extensions_path=invalid_name_path)
563 def test_extensions_column_id_auto_generation(self) -> None:
564 auto_id_path = os.path.join(self.tmpdir, "auto_id.yaml")
565 with open(auto_id_path, "w") as f:
566 f.write("""
567name: auto_id
568tables:
569 - name: schemas
570 columns:
571 - name: auto_id
572 datatype: int
573 nullable: false
574 description: "Column with auto_id"
575""")
577 mgr = TableManager(extensions_path=auto_id_path)
578 schemas_table = mgr["schemas"]
579 self.assertIn("auto_id", schemas_table.c)
581 def test_extensions_column_id_preserved(self) -> None:
582 explicit_id_path = os.path.join(self.tmpdir, "explicit_id.yaml")
583 with open(explicit_id_path, "w") as f:
584 f.write("""
585name: explicit_id
586tables:
587 - name: schemas
588 columns:
589 - name: explicit_id
590 datatype: int
591 "@id": "#custom.id.path"
592""")
594 mgr = TableManager(extensions_path=explicit_id_path)
595 schemas_table = mgr["schemas"]
596 self.assertIn("explicit_id", schemas_table.c)
598 def test_extensions_multiple_tables_extended(self) -> None:
599 multi_table_path = os.path.join(self.tmpdir, "multi_table.yaml")
600 with open(multi_table_path, "w") as f:
601 f.write("""
602name: multi_table
603tables:
604 - name: schemas
605 columns:
606 - name: schema_ext1
607 datatype: int
608 - name: schema_ext2
609 datatype: int
610 - name: tables
611 columns:
612 - name: table_ext1
613 datatype: int
614 - name: table_ext2
615 datatype: double
616 - name: columns
617 columns:
618 - name: col_ext1
619 datatype: int
620 - name: keys
621 columns:
622 - name: key_ext1
623 datatype: char
624 length: 128
625""")
627 mgr = TableManager(extensions_path=multi_table_path)
629 schemas_table = mgr["schemas"]
630 self.assertIn("schema_ext1", schemas_table.c)
631 self.assertIn("schema_ext2", schemas_table.c)
633 tables_table = mgr["tables"]
634 self.assertIn("table_ext1", tables_table.c)
635 self.assertIn("table_ext2", tables_table.c)
637 columns_table = mgr["columns"]
638 self.assertIn("col_ext1", columns_table.c)
640 keys_table = mgr["keys"]
641 self.assertIn("key_ext1", keys_table.c)
643 def test_extensions_nonexistent_table_skipped(self) -> None:
644 nonexistent_table_path = os.path.join(self.tmpdir, "nonexistent_table.yaml")
645 with open(nonexistent_table_path, "w") as f:
646 f.write("""
647name: test_extensions_nonexistent_table
648tables:
649 - name: schemas
650 columns:
651 - name: valid_ext
652 datatype: int
653 - name: nonexistent_table
654 columns:
655 - name: should_be_ignored
656 datatype: int
657""")
659 mgr = TableManager(extensions_path=nonexistent_table_path)
660 schemas_table = mgr["schemas"]
661 self.assertIn("valid_ext", schemas_table.c)
663 def test_extensions_column_properties_preserved(self) -> None:
664 full_props_path = os.path.join(self.tmpdir, "full_props.yaml")
665 with open(full_props_path, "w") as f:
666 f.write("""
667name: full_props
668tables:
669 - name: schemas
670 columns:
671 - name: full_property_column
672 datatype: char
673 length: 64
674 nullable: false
675 description: "Column with all properties"
676 "@id": "#tap_schema.schemas.full_property_column"
677""")
679 mgr = TableManager(extensions_path=full_props_path)
680 schemas_table = mgr["schemas"]
681 self.assertIn("full_property_column", schemas_table.c)
683 def test_extensions_apply_schema_to_metadata_true(self) -> None:
684 mgr = TableManager(apply_schema_to_metadata=True, extensions_path=self.extensions_path)
685 schemas_table = mgr["schemas"]
686 self.assertIn("owner_id", schemas_table.c)
688 def test_extensions_apply_schema_to_metadata_false(self) -> None:
689 mgr = TableManager(apply_schema_to_metadata=False, extensions_path=self.extensions_path)
691 schemas_table = mgr["schemas"]
692 self.assertIn("owner_id", schemas_table.c)
693 self.assertIn("read_anon", schemas_table.c)
695 def test_extensions_with_table_name_postfix(self) -> None:
696 mgr = TableManager(
697 apply_schema_to_metadata=False, extensions_path=self.extensions_path, table_name_postfix="_custom"
698 )
700 schemas_table = mgr["schemas"]
701 self.assertIn("owner_id", schemas_table.c)
703 def test_extensions_metadata_builder_called(self) -> None:
704 mgr = TableManager(extensions_path=self.extensions_path)
706 self.assertIsNotNone(mgr._metadata)
708 table_names = list(mgr.metadata.tables.keys())
709 found_schemas = any("schemas" in name for name in table_names)
710 found_tables = any("tables" in name and "schemas" not in name for name in table_names)
712 self.assertTrue(found_schemas, f"No schemas table found in {table_names}")
713 self.assertTrue(found_tables, f"No tables table found in {table_names}")
715 def test_extensions_preserve_original_columns(self) -> None:
716 mgr = TableManager(extensions_path=self.extensions_path)
718 schemas_table = mgr["schemas"]
719 column_names = [col.name for col in schemas_table.columns]
721 self.assertIn("schema_name", column_names)
722 self.assertIn("owner_id", column_names)
723 self.assertIn("read_anon", column_names)
725 def test_no_extensions_path_provided(self) -> None:
726 mgr = TableManager(extensions_path=None)
727 schemas_table = mgr["schemas"]
728 self.assertNotIn("owner_id", schemas_table.c)
730 def test_extensions_path_empty_string(self) -> None:
731 mgr = TableManager(extensions_path="")
732 schemas_table = mgr["schemas"]
733 self.assertNotIn("owner_id", schemas_table.c)
735 def test_extensions_file_not_found(self) -> None:
736 nonexistent_path = os.path.join(self.tmpdir, "does_not_exist.yaml")
737 with self.assertRaises(ValueError):
738 TableManager(extensions_path=nonexistent_path)
741if __name__ == "__main__": 741 ↛ 742line 741 didn't jump to line 742 because the condition on line 741 was never true
742 unittest.main()