From abd11d1307ae7e457c8847bbed2d51e12069a2c7 Mon Sep 17 00:00:00 2001 From: Jeremy McCormick Date: Mon, 19 Aug 2024 17:48:20 -0500 Subject: [PATCH] test_tap_schema: Add simple test of loading TAP_SCHEMA self-description schema --- tests/test_tap_schema.py | 33 ++++++++++++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/tests/test_tap_schema.py b/tests/test_tap_schema.py index bde83256..73a8d212 100644 --- a/tests/test_tap_schema.py +++ b/tests/test_tap_schema.py @@ -26,7 +26,7 @@ from sqlalchemy import MetaData, create_engine, select from felis.datamodel import Schema -from felis.tap_schema import DataLoader, TableManager +from felis.tap_schema import _TAP_SCHEMA_YAML, DataLoader, TableManager TESTDIR = os.path.abspath(os.path.dirname(__file__)) TEST_YAML = os.path.join(TESTDIR, "data", "sales.yaml") @@ -198,3 +198,34 @@ def test_key_columns(self) -> None: self.assertEqual(key_column["key_id"], "fk_table1_to_table2") self.assertEqual(key_column["from_column"], "fk") self.assertEqual(key_column["target_column"], "id") + + +class TapSchemaMetaTest(unittest.TestCase): + """Test loading of a TAP_SCHEMA representation from a YAML schema into + TAP_SCHEMA itself. + """ + + def setUp(self) -> None: + """Set up the test case.""" + self.schema = Schema.model_validate( + yaml.safe_load(open(_TAP_SCHEMA_YAML)), context={"generate_ids": True} + ) + + self.engine = create_engine("sqlite:///:memory:") + + mgr = TableManager(apply_schema_to_metadata=False, table_name_postfix=TABLE_NAME_POSTFIX) + mgr.initialize_database(self.engine) + self.mgr = mgr + + loader = DataLoader(self.schema, mgr, self.engine, outfile=None) + loader.load() + + self.md = MetaData() + self.md.reflect(self.engine) + + def test_tap_schema(self) -> None: + with self.engine.connect() as connection: + for table_name in ["schemas11", "tables11", "columns11", "keys11", "key_columns11"]: + result = connection.execute(select(self.md.tables[table_name])) + for row in result: + print(row)