Check if package metadata matches Alembic migrations

test/unit/io_managers_test.py

@@ -1,7 +1,9 @@
 """Test Dagster IO Managers."""
 import datetime
 import json
+from pathlib import Path
 
+import alembic.config
 import hypothesis
 import pandas as pd
 import pandera
@@ -204,40 +206,53 @@ def test_missing_schema_error(sqlite_io_manager_fixture):
 
 
 @pytest.fixture
-def pudl_sqlite_io_manager_fixture(tmp_path, test_pkg):
-    """Create a SQLiteIOManager fixture with a PUDL database schema."""
-    db_path = tmp_path / "pudl.sqlite"
+def fake_pudl_sqlite_io_manager_fixture(tmp_path, test_pkg, monkeypatch):
+    """Create a SQLiteIOManager fixture with a fake database schema."""
+    db_path = tmp_path / "fake.sqlite"
 
     # Create the database and schemas
     engine = sa.create_engine(f"sqlite:///{db_path}")
     md = test_pkg.to_sql()
     md.create_all(engine)
-    return PudlSQLiteIOManager(base_dir=tmp_path, db_name="pudl", package=test_pkg)
+    return PudlSQLiteIOManager(base_dir=tmp_path, db_name="fake", package=test_pkg)
 
 
-def test_error_when_handling_view_without_metadata(pudl_sqlite_io_manager_fixture):
+def test_migrations_match_metadata(tmp_path, monkeypatch):
+    pkg = Package.from_resource_ids()
+    monkeypatch.chdir(Path(__file__).parent.parent.parent)
+    # alembic knows to use PudlPaths().pudl_db - so we need to set PUDL_OUTPUT env var
+    monkeypatch.setenv("PUDL_OUTPUT", tmp_path)
+    alembic.config.main(["upgrade", "head"])
+
+    PudlSQLiteIOManager(base_dir=tmp_path, db_name="pudl", package=pkg)
+
+    # all we care about is that it didn't raise an error
+    assert True
+
+
+def test_error_when_handling_view_without_metadata(fake_pudl_sqlite_io_manager_fixture):
     """Make sure an error is thrown when a user creates a view without metadata."""
     asset_key = "track_view"
     sql_stmt = "CREATE VIEW track_view AS SELECT * FROM track;"
     output_context = build_output_context(asset_key=AssetKey(asset_key))
     with pytest.raises(ValueError):
-        pudl_sqlite_io_manager_fixture.handle_output(output_context, sql_stmt)
+        fake_pudl_sqlite_io_manager_fixture.handle_output(output_context, sql_stmt)
 
 
 @pytest.mark.skip(reason="SQLAlchemy is not finding the view. Debug or remove.")
-def test_handling_view_with_metadata(pudl_sqlite_io_manager_fixture):
+def test_handling_view_with_metadata(fake_pudl_sqlite_io_manager_fixture):
     """Make sure an users can create and load views when it has metadata."""
     # Create some sample data
     asset_key = "artist"
     artist = pd.DataFrame({"artistid": [1], "artistname": ["Co-op Mop"]})
     output_context = build_output_context(asset_key=AssetKey(asset_key))
-    pudl_sqlite_io_manager_fixture.handle_output(output_context, artist)
+    fake_pudl_sqlite_io_manager_fixture.handle_output(output_context, artist)
 
     # create the view
     asset_key = "artist_view"
     sql_stmt = "CREATE VIEW artist_view AS SELECT * FROM artist;"
     output_context = build_output_context(asset_key=AssetKey(asset_key))
-    pudl_sqlite_io_manager_fixture.handle_output(output_context, sql_stmt)
+    fake_pudl_sqlite_io_manager_fixture.handle_output(output_context, sql_stmt)
 
     # read the view data as a dataframe
     input_context = build_input_context(asset_key=AssetKey(asset_key))
@@ -246,15 +261,15 @@ def test_handling_view_with_metadata(pudl_sqlite_io_manager_fixture):
     # sqlalchemy.exc.InvalidRequestError: Could not reflect: requested table(s) not available in
     # Engine(sqlite:////private/var/folders/pg/zrqnq8l113q57bndc5__h2640000gn/
     # # T/pytest-of-nelsonauner/pytest-38/test_handling_view_with_metada0/pudl.sqlite): (artist_view)
-    pudl_sqlite_io_manager_fixture.load_input(input_context)
+    fake_pudl_sqlite_io_manager_fixture.load_input(input_context)
 
 
-def test_error_when_reading_view_without_metadata(pudl_sqlite_io_manager_fixture):
+def test_error_when_reading_view_without_metadata(fake_pudl_sqlite_io_manager_fixture):
     """Make sure and error is thrown when a user loads a view without metadata."""
     asset_key = "track_view"
     input_context = build_input_context(asset_key=AssetKey(asset_key))
     with pytest.raises(ValueError):
-        pudl_sqlite_io_manager_fixture.load_input(input_context)
+        fake_pudl_sqlite_io_manager_fixture.load_input(input_context)
 
 
 def test_ferc_xbrl_sqlite_io_manager_dedupes(mocker, tmp_path):