Skip to content

Commit

Permalink
test_tap_schema: Add simple test of loading TAP_SCHEMA self-descripti…
Browse files Browse the repository at this point in the history
…on schema
  • Loading branch information
JeremyMcCormick committed Aug 19, 2024
1 parent d294eef commit abd11d1
Showing 1 changed file with 32 additions and 1 deletion.
33 changes: 32 additions & 1 deletion tests/test_tap_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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)

0 comments on commit abd11d1

Please sign in to comment.