Source code for message_ix_models.tests.model.transport.test_util

import pandas as pd
import pandas.testing as pdt
import pytest
import xarray as xr
from genno.testing import assert_qty_equal
from iam_units import registry

from message_ix_models.model.transport.config import Config, DataSourceConfig
from message_ix_models.report.util import as_quantity


[docs] @pytest.mark.xfail(reason="Refactoring") def test_add_cl(test_context): """:func:`.input_commodity_level` preserves the content of other columns.""" from message_ix_models.model.transport.util import input_commodity_level # Input data missing 'commodity' and 'level' df_in = pd.DataFrame( [ ["R11_AFR", None, None, "ICE_conv"], ["R11_WEU", None, None, "ELC_100"], ], columns=["node", "commodity", "level", "technology"], ) Config.from_context(test_context) df_out = input_commodity_level(test_context, df_in, default_level="foo") # Output is the same shape assert df_out.shape == (2, 4) # All NaNs are filled assert not df_out.isna().any().any(), df_out # Existing columns have the same content for col in "node", "technology": pdt.assert_series_equal(df_in[col], df_out[col])
[docs] @pytest.mark.parametrize( "regions", [ None, # Default per message_ix_models.model.Config "R11", "R12", "R14", pytest.param("ISR", marks=pytest.mark.xfail(raises=AssertionError)), ], ) def test_configure(test_context, regions): """Configuration can be read from files. This exercises :meth:`.Config.from_context`. """ # Set the regional aggregation to be used ctx = test_context if regions: ctx.model.regions = regions # Returns the same object stored as Context["transport"] cfg = Config.from_context(ctx) assert cfg is ctx["transport"] # Attributes have the correct types assert isinstance(cfg.data_source, DataSourceConfig) # Scalar parameters are loaded assert cfg.scaling assert_qty_equal(as_quantity("200 * 8 hours / passenger / year"), cfg.work_hours) # Codes for the consumer_group set are generated codes = cfg.spec.add.set["consumer_group"] RUEAA = codes[codes.index("RUEAA")] assert "Rural, or “Outside MSA”, Early Adopter, Average" == str(RUEAA.name) # xarray objects are generated for advanced indexing indexers = cfg.spec.add.set["consumer_group indexers"] assert all(isinstance(da, xr.DataArray) for da in indexers.values()) # Codes for commodities are generated codes = cfg.spec.add.set["commodity"] RUEAA = codes[codes.index("transport pax RUEAA")] assert RUEAA.eval_annotation("demand") is True # …with expected units r = dict(registry=registry) assert registry.Unit("Gp km") == RUEAA.eval_annotation("units", r) # Codes for technologies are generated, with annotations giving their units codes = cfg.spec.add.set["technology"] ELC_100 = codes[codes.index("ELC_100")] assert registry.Unit("Gv km") == ELC_100.eval_annotation("units", r) # If "ISR" was given as 'regions', then the corresponding config file was loaded if regions == "ISR": # Check one config value to confirm assert {"Israel"} == set(cfg.node_to_census_division.keys())
[docs] @pytest.mark.parametrize( "options", [ {}, pytest.param( {"mode-share": "default"}, marks=pytest.mark.xfail(raises=TypeError) ), {"mode_share": "default"}, {"mode_share": "INVALID"}, {"futures_scenario": "base"}, {"futures_scenario": "A---"}, {"futures_scenario": "debug"}, ], ) def test_configure_options(test_context, options): """:func:`.configure` operates with various options.""" Config.from_context(test_context, options=options)