diff --git a/test/unit/io_managers_test.py b/test/unit/io_managers_test.py index 89d8a80988..d9c87885bf 100644 --- a/test/unit/io_managers_test.py +++ b/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):