diff --git a/pixi.toml b/pixi.toml index 20fb387..50c2a9f 100644 --- a/pixi.toml +++ b/pixi.toml @@ -12,7 +12,7 @@ workflow_test = """python workflow.py --config pypsa_validation_processing/confi python workflow.py --config pypsa_validation_processing/configs/config.region-timeseries.yaml&& python workflow.py --config pypsa_validation_processing/configs/config.region-year.yaml&& pytest tests/ -v""" -test = "pytest tests/ -v" +test = "pytest tests/ -v --cov=pypsa_validation_processing --cov-report=term-missing --cov-fail-under=90" [dependencies] python = ">=3.11" diff --git a/pypsa_validation_processing/class_definitions.py b/pypsa_validation_processing/class_definitions.py index cd8c599..dfe36c3 100644 --- a/pypsa_validation_processing/class_definitions.py +++ b/pypsa_validation_processing/class_definitions.py @@ -686,6 +686,10 @@ def _get_unit_from_common_definitions(self, variable: str) -> str: ) target_unit = match.iloc[0][unit_col] + if pd.isna(target_unit): + raise KeyError( + f"Unit information not found for variable '{variable}' in common definitions." + ) if ("[" in target_unit) and ("]" in target_unit) and ("," in target_unit): print( "Several possible units defined for variable '{variable}' in common definitions. Take first one:".format( @@ -702,10 +706,6 @@ def _get_unit_from_common_definitions(self, variable: str) -> str: f"WARNING: Failed to parse multiple units for variable '{variable}': {exc}. Using TJ as unit." ) target_unit = "TJ" - if pd.isna(target_unit): - raise KeyError( - f"Unit information not found for variable '{variable}' in common definitions." - ) return str(target_unit) def _get_network_config(self, investment_year): diff --git a/tests/test_format_timestamps.py b/tests/test_format_timestamps.py index 18382f1..9e23474 100644 --- a/tests/test_format_timestamps.py +++ b/tests/test_format_timestamps.py @@ -110,6 +110,26 @@ def tz_localize(self, _tz): assert "WARNING: format_timestamps: failed to localize column" in captured.out +def test_format_timestamps_falls_back_when_mixed_format_raises_typeerror(): + real_to_datetime = pd.to_datetime + + def fake_to_datetime(cols, errors=None, format=None): + if format == "mixed": + raise TypeError("format='mixed' not supported") + return real_to_datetime(cols, errors=errors) + + df = pd.DataFrame([[1.0]], columns=["2050-01-01 00:00:00"]) + + with patch( + "pypsa_validation_processing.class_definitions.pd.to_datetime", + side_effect=fake_to_datetime, + ): + out = format_timestamps(df) + + assert out.columns[0].year == 2050 + assert out.columns[0].utcoffset() == datetime.timedelta(hours=0) + + def test_structure_pyam_from_pandas_formats_time_columns_before_pyam(tmp_path: Path): processor = _make_processor(tmp_path) processor.aggregation_level = "country" diff --git a/tests/test_network_processor.py b/tests/test_network_processor.py index e426cae..7c3baf5 100644 --- a/tests/test_network_processor.py +++ b/tests/test_network_processor.py @@ -117,6 +117,114 @@ def test_repr_method(self, mock_config_file: Path): assert "Network_Processor" in repr_str assert "AT" in repr_str + def test_init_nonexistent_network_results_path_raises( + self, tmp_path: Path, mock_definitions_path: Path + ): + """Test that a missing network_results_path raises FileNotFoundError.""" + missing_nw_path = tmp_path / "does_not_exist" + config_content = f""" +country: AT +model_name: AT_KN2040 +scenario_name: test_scenario +definitions_path: {mock_definitions_path} +network_results_path: {missing_nw_path} +""" + config_file = tmp_path / "config.yaml" + config_file.write_text(config_content) + + with pytest.raises(FileNotFoundError, match="Network results folder"): + Network_Processor(config_path=config_file) + + def test_init_missing_definitions_path_key_raises_value_error( + self, tmp_path: Path, mock_network_results_path: Path + ): + """Test that an unset definitions_path raises ValueError.""" + config_content = f""" +country: AT +model_name: AT_KN2040 +scenario_name: test_scenario +network_results_path: {mock_network_results_path} +""" + config_file = tmp_path / "config.yaml" + config_file.write_text(config_content) + + with pytest.raises(ValueError, match="definition_path"): + Network_Processor(config_path=config_file) + + def test_init_nonexistent_definitions_path_raises( + self, tmp_path: Path, mock_network_results_path: Path + ): + """Test that a missing definitions_path directory raises FileNotFoundError.""" + missing_defs_path = tmp_path / "no_definitions_here" + config_content = f""" +country: AT +model_name: AT_KN2040 +scenario_name: test_scenario +definitions_path: {missing_defs_path} +network_results_path: {mock_network_results_path} +""" + config_file = tmp_path / "config.yaml" + config_file.write_text(config_content) + + with pytest.raises(FileNotFoundError, match="Definition folder"): + Network_Processor(config_path=config_file) + + def test_init_null_model_name_raises_value_error( + self, + tmp_path: Path, + mock_definitions_path: Path, + mock_network_results_path: Path, + ): + """Test that a null model_name in config raises ValueError.""" + config_content = f""" +country: AT +model_name: +scenario_name: test_scenario +definitions_path: {mock_definitions_path} +network_results_path: {mock_network_results_path} +""" + config_file = tmp_path / "config.yaml" + config_file.write_text(config_content) + + with patch( + "pypsa_validation_processing.class_definitions.pypsa.NetworkCollection" + ): + with patch( + "pypsa_validation_processing.class_definitions.nomenclature.DataStructureDefinition" + ): + with pytest.raises( + ValueError, match="model_name.*scenario_name.*must be set" + ): + Network_Processor(config_path=config_file) + + def test_init_null_scenario_name_raises_value_error( + self, + tmp_path: Path, + mock_definitions_path: Path, + mock_network_results_path: Path, + ): + """Test that a null scenario_name in config raises ValueError.""" + config_content = f""" +country: AT +model_name: AT_KN2040 +scenario_name: +definitions_path: {mock_definitions_path} +network_results_path: {mock_network_results_path} +""" + config_file = tmp_path / "config.yaml" + config_file.write_text(config_content) + + with patch( + "pypsa_validation_processing.class_definitions.pypsa.NetworkCollection" + ): + with patch( + "pypsa_validation_processing.class_definitions.nomenclature.DataStructureDefinition" + ): + with pytest.raises( + ValueError, match="model_name.*scenario_name.*must be set" + ): + Network_Processor(config_path=config_file) + # --------------------------------------------------------------------------- # Tests for configuration reading @@ -192,6 +300,31 @@ def test_execute_function_not_found(self, mock_config_file: Path): ) assert result is None + def test_execute_function_warns_when_function_name_not_in_module( + self, mock_config_file: Path, capsys + ): + """Test that a WARNING is printed and None returned for an unresolvable function name.""" + with patch( + "pypsa_validation_processing.class_definitions.pypsa.NetworkCollection" + ): + with patch( + "pypsa_validation_processing.class_definitions.nomenclature.DataStructureDefinition" + ): + processor = Network_Processor(config_path=mock_config_file) + processor.functions_dict = { + "Bogus Variable": "This_Function_Does_Not_Exist" + } + + mock_network = MockPyPSANetwork() + result = processor._execute_function_for_variable( + "Bogus Variable", mock_network + ) + + captured = capsys.readouterr() + assert result is None + assert "WARNING" in captured.out + assert "This_Function_Does_Not_Exist" in captured.out + def test_execute_function_passes_config_when_accepted(self, mock_config_file: Path): """Test that config is passed to functions that accept it.""" with patch( @@ -1409,3 +1542,185 @@ def test_sanitize_path_token_edge_cases(self): assert Network_Processor._sanitize_path_token("") == "" assert Network_Processor._sanitize_path_token(" ") == "" assert Network_Processor._sanitize_path_token("NoWhitespace") == "NoWhitespace" + + +# --------------------------------------------------------------------------- +# Tests for _get_network_config +# --------------------------------------------------------------------------- + + +class TestGetNetworkConfig: + """Test _get_network_config() config-file discovery and loading.""" + + def _setup_processor(self, tmp_path: Path) -> Network_Processor: + defs_path = tmp_path / "definitions" + defs_path.mkdir(exist_ok=True) + nw_path = tmp_path / "networks" + nw_path.mkdir(parents=True, exist_ok=True) + (nw_path / "dummy.nc").touch() + config_content = f""" +country: AT +model_name: test_model +scenario_name: test_scenario +definitions_path: {defs_path} +network_results_path: {tmp_path} +output_path: {tmp_path / 'output.xlsx'} +""" + config_file = tmp_path / "config.yaml" + config_file.write_text(config_content) + + with patch( + "pypsa_validation_processing.class_definitions.pypsa.NetworkCollection" + ): + with patch( + "pypsa_validation_processing.class_definitions.nomenclature.DataStructureDefinition" + ): + return Network_Processor(config_path=config_file) + + def test_warns_and_uses_first_of_multiple_matching_config_files( + self, tmp_path: Path, capsys + ): + """Test that multiple matching config files trigger an INFO message and use the first.""" + processor = self._setup_processor(tmp_path) + configs_dir = processor.network_results_path / "configs" + configs_dir.mkdir(parents=True, exist_ok=True) + (configs_dir / "config_a_2020.yaml").write_text("foo: bar\n") + (configs_dir / "config_b_2020.yaml").write_text("foo: baz\n") + + result = processor._get_network_config(2020) + + captured = capsys.readouterr() + assert "INFO: Multiple config files found" in captured.out + assert result is not None + + def test_warns_and_returns_none_on_malformed_yaml(self, tmp_path: Path, capsys): + """Test that malformed YAML in the matched config file triggers a WARNING.""" + processor = self._setup_processor(tmp_path) + configs_dir = processor.network_results_path / "configs" + configs_dir.mkdir(parents=True, exist_ok=True) + (configs_dir / "config_2020.yaml").write_text("foo: [unclosed\n") + + result = processor._get_network_config(2020) + + captured = capsys.readouterr() + assert "WARNING: Could not load config file" in captured.out + assert result is None + + def test_warns_and_returns_none_when_no_config_file_found( + self, tmp_path: Path, capsys + ): + """Test that no matching config file triggers a WARNING and returns None.""" + processor = self._setup_processor(tmp_path) + + result = processor._get_network_config(2020) + + captured = capsys.readouterr() + assert "WARNING: No config file found" in captured.out + assert result is None + + +# --------------------------------------------------------------------------- +# Tests for calculate_variables_values aggregation branches +# --------------------------------------------------------------------------- + + +class TestCalculateVariablesValuesAggregation: + """Test calculate_variables_values() for aggregate_per_year branches.""" + + def _setup_processor(self, tmp_path: Path) -> Network_Processor: + defs_path = tmp_path / "definitions" + defs_path.mkdir(exist_ok=True) + nw_path = tmp_path / "networks" + nw_path.mkdir(parents=True, exist_ok=True) + (nw_path / "dummy.nc").touch() + config_content = f""" +country: AT +model_name: test_model +scenario_name: test_scenario +definitions_path: {defs_path} +network_results_path: {tmp_path} +output_path: {tmp_path / 'output.xlsx'} +""" + config_file = tmp_path / "config.yaml" + config_file.write_text(config_content) + + with patch( + "pypsa_validation_processing.class_definitions.pypsa.NetworkCollection" + ): + with patch( + "pypsa_validation_processing.class_definitions.nomenclature.DataStructureDefinition" + ): + return Network_Processor(config_path=config_file) + + def test_raises_runtime_error_when_result_not_dataframe_and_not_aggregated( + self, tmp_path: Path + ): + """Test RuntimeError when aggregate_per_year=False and result isn't a DataFrame.""" + processor = self._setup_processor(tmp_path) + processor.aggregate_per_year = False + processor.aggregation_level = "region" + + network = MockPyPSANetwork() + processor.network_collection = MockNetworkCollection([network]) + + processor.dsd = MagicMock() + processor.dsd.variable.to_pandas.return_value = pd.DataFrame( + {"variable": ["Test Variable"]} + ) + + not_a_dataframe = pd.Series( + [1.0], + index=pd.MultiIndex.from_tuples( + [("AT1", "MWh_el")], names=["location", "unit"] + ), + ) + + with patch.object( + processor, + "_execute_function_for_variable", + return_value=not_a_dataframe, + ): + with pytest.raises(RuntimeError, match="Expected DataFrame"): + processor.calculate_variables_values() + + def test_merges_multiple_investment_years_when_aggregated(self, tmp_path: Path): + """calculate_variables_values merges per-year results into one wide table.""" + processor = self._setup_processor(tmp_path) + processor.aggregate_per_year = True + processor.aggregation_level = "region" + + network_2020 = MockPyPSANetwork(name="network_2020") + network_2020.meta["wildcards"]["planning_horizons"] = 2020 + network_2030 = MockPyPSANetwork(name="network_2030") + network_2030.meta["wildcards"]["planning_horizons"] = 2030 + processor.network_collection = MockNetworkCollection( + [network_2020, network_2030] + ) + + processor.dsd = MagicMock() + processor.dsd.variable.to_pandas.return_value = pd.DataFrame( + {"variable": ["Test Variable"]} + ) + + def fake_execute(variable, n, config=None): + year = n.meta["wildcards"]["planning_horizons"] + return pd.Series( + [float(year)], + index=pd.MultiIndex.from_tuples( + [("AT1", "MWh_el")], names=["location", "unit"] + ), + ) + + with patch.object( + processor, "_execute_function_for_variable", side_effect=fake_execute + ): + with patch.object( + processor, "structure_pyam_from_pandas", side_effect=lambda df: df + ): + processor.calculate_variables_values() + + result = processor.dsd_with_values + assert isinstance(result, pd.DataFrame) + assert "2020" in result.columns + assert "2030" in result.columns + assert len(result) == 1 diff --git a/tests/test_no_duplicate_test_classes.py b/tests/test_no_duplicate_test_classes.py new file mode 100644 index 0000000..b1be917 --- /dev/null +++ b/tests/test_no_duplicate_test_classes.py @@ -0,0 +1,33 @@ +"""Guard against silently shadowed test classes. + +Python allows redefining a top-level class within the same module; the second +definition silently overwrites the first and pytest collects only the survivor, +with no warning. This test walks every module in ``tests/`` and fails if any +top-level class name is defined more than once. +""" + +import ast +import collections +from pathlib import Path + +import pytest + +TESTS_DIR = Path(__file__).parent + + +def _duplicate_top_level_classes(path: Path) -> list[str]: + tree = ast.parse(path.read_text(), filename=str(path)) + names = [node.name for node in tree.body if isinstance(node, ast.ClassDef)] + counts = collections.Counter(names) + return sorted(name for name, count in counts.items() if count > 1) + + +@pytest.mark.parametrize( + "path", sorted(TESTS_DIR.glob("test_*.py")), ids=lambda p: p.name +) +def test_no_duplicate_top_level_class_names(path: Path): + duplicates = _duplicate_top_level_classes(path) + assert not duplicates, ( + f"{path.name} defines the following top-level class(es) more than once, " + f"which silently shadows earlier test methods: {duplicates}" + ) diff --git a/tests/test_statistics_functions.py b/tests/test_statistics_functions.py index b748e0f..893e975 100644 --- a/tests/test_statistics_functions.py +++ b/tests/test_statistics_functions.py @@ -1333,284 +1333,6 @@ def test_returns_dataframe_for_aggregate_per_year_false(self): pd.testing.assert_index_equal(result.columns, network.snapshots) -# --------------------------------------------------------------------------- -# Tests for Final_Energy_by_Carrier__Oil -# --------------------------------------------------------------------------- - - -class TestFinalEnergyByCarrierOil: - """Test suite for Final_Energy_by_Carrier__Oil function.""" - - class _OilStatisticsAccessor: - """Deterministic accessor tailored to oil final-energy tests.""" - - def __init__( - self, - *, - rescom_empty: bool = False, - all_oil_value: float = 200.0, - all_oil_empty: bool = False, - non_fossil_empty: bool = False, - ): - self.rescom_empty = rescom_empty - self.all_oil_value = all_oil_value - self.all_oil_empty = all_oil_empty - self.non_fossil_empty = non_fossil_empty - - def _to_result( - self, - *, - index: pd.MultiIndex, - values: list[float], - groupby_time: bool, - ) -> pd.Series | pd.DataFrame: - if groupby_time: - return pd.Series(values, index=index, dtype=float) - timestamps = pd.date_range( - "2019-01-01", periods=4, freq="6h", name="snapshot" - ) - return pd.DataFrame( - {ts: values for ts in timestamps}, index=index, dtype=float - ) - - def withdrawal( - self, - bus_carrier: str | None = None, - carrier: list[str] | str | None = None, - components: str | list[str] | None = None, - groupby_time: bool = True, - groupby: list[str] | None = None, - at_port: str | None = None, - **kwargs: object, - ) -> pd.Series | pd.DataFrame: - if groupby is None: - groupby = ["location", "unit"] - - # Agriculture and land-transport final demand (Load) - if carrier == "agriculture machinery oil" and components == "Load": - idx = pd.MultiIndex.from_tuples( - [("AT1", "MWh_th")], names=["location", "unit"] - ) - return self._to_result( - index=idx, values=[100.0], groupby_time=groupby_time - ) - - if carrier == "land transport oil" and components == "Load": - idx = pd.MultiIndex.from_tuples( - [("AT1", "MWh_th")], names=["location", "unit"] - ) - return self._to_result( - index=idx, values=[300.0], groupby_time=groupby_time - ) - - # Residential/commercial demand requiring copperplate -> location mapping via bus1 - if ( - bus_carrier == "oil" - and isinstance(carrier, list) - and set(carrier) == {"rural oil boiler", "urban decentral oil boiler"} - ): - idx_names = ["name", "bus", "carrier", "location", "unit", "bus1"] - if self.rescom_empty: - empty_idx = pd.MultiIndex.from_arrays( - [[] for _ in idx_names], names=idx_names - ) - return self._to_result( - index=empty_idx, - values=[], - groupby_time=groupby_time, - ) - idx = pd.MultiIndex.from_tuples( - [ - ( - "rural_boiler_load", - "AT1 oil", - "rural oil boiler", - "EU", - "MWh_th", - "AT1 oil", - ), - ( - "urban_boiler_load", - "AT1 oil", - "urban decentral oil boiler", - "EU", - "MWh_th", - "AT1 oil", - ), - ], - names=idx_names, - ) - return self._to_result( - index=idx, - values=[50.0, 50.0], - groupby_time=groupby_time, - ) - - # Total oil use denominator for non-fossil share - if ( - bus_carrier == "oil" - and components == "Link" - and at_port == "bus0" - and groupby == ["bus1", "carrier", "location", "unit"] - ): - if self.all_oil_empty: - empty_idx = pd.MultiIndex.from_arrays( - [[], [], [], []], - names=["bus1", "carrier", "location", "unit"], - ) - return self._to_result( - index=empty_idx, - values=[], - groupby_time=groupby_time, - ) - idx = pd.MultiIndex.from_tuples( - [ - ( - "AT1 oil", - "land transport oil", - "EU", - "MWh_th", - ) - ], - names=["bus1", "carrier", "location", "unit"], - ) - return self._to_result( - index=idx, - values=[self.all_oil_value], - groupby_time=groupby_time, - ) - - raise AssertionError( - f"Unexpected withdrawal call: bus_carrier={bus_carrier}, carrier={carrier}, components={components}, groupby={groupby}, at_port={at_port}" - ) - - def supply( - self, - bus_carrier: str | None = None, - carrier: list[str] | str | None = None, - at_port: str | None = None, - components: str | list[str] | None = None, - groupby: list[str] | None = None, - groupby_time: bool = True, - **kwargs: object, - ) -> pd.Series | pd.DataFrame: - if ( - bus_carrier == "oil" - and components == "Link" - and at_port == "bus1" - and groupby == ["name", "bus", "carrier", "location", "unit", "bus0"] - ): - if self.non_fossil_empty: - empty_idx = pd.MultiIndex.from_arrays( - [[], [], [], [], [], []], - names=["name", "bus", "carrier", "location", "unit", "bus0"], - ) - return self._to_result( - index=empty_idx, - values=[], - groupby_time=groupby_time, - ) - idx = pd.MultiIndex.from_tuples( - [ - ( - "renewable_oil_link", - "AT1 oil", - "biomass to liquid", - "EU", - "MWh_th", - "AT1 oil", - ) - ], - names=["name", "bus", "carrier", "location", "unit", "bus0"], - ) - return self._to_result( - index=idx, values=[500.0], groupby_time=groupby_time - ) - - raise AssertionError( - f"Unexpected supply call: bus_carrier={bus_carrier}, carrier={carrier}, components={components}, groupby={groupby}, at_port={at_port}" - ) - - class _OilNetwork: - """Minimal network object exposing only the statistics accessor.""" - - def __init__( - self, - *, - rescom_empty: bool = False, - all_oil_value: float = 200.0, - all_oil_empty: bool = False, - non_fossil_empty: bool = False, - ): - self.statistics = TestFinalEnergyByCarrierOil._OilStatisticsAccessor( - rescom_empty=rescom_empty, - all_oil_value=all_oil_value, - all_oil_empty=all_oil_empty, - non_fossil_empty=non_fossil_empty, - ) - - def test_clips_non_fossil_share_above_one_to_zero_fossil(self): - """Renewable oil production above total demand should yield zero fossil oil.""" - result = Final_Energy_by_Carrier__Oil(self._OilNetwork()) - - assert isinstance(result, pd.Series) - assert isinstance(result.index, pd.MultiIndex) - assert result.index.names == ["location", "unit"] - assert result.loc[("AT1", "MWh")] == pytest.approx(0.0) - - def test_handles_empty_rescom_without_failing(self): - """Function should work even when residential/commercial oil demand is empty.""" - result = Final_Energy_by_Carrier__Oil(self._OilNetwork(rescom_empty=True)) - - assert isinstance(result, pd.Series) - assert isinstance(result.index, pd.MultiIndex) - assert result.index.names == ["location", "unit"] - # non-fossil fraction is clipped to 1, so fossil share remains zero. - assert result.loc[("AT1", "MWh")] == pytest.approx(0.0) - - def test_handles_zero_total_oil_demand_denominator(self): - """Division by zero in non-fossil share denominator should not crash.""" - result = Final_Energy_by_Carrier__Oil(self._OilNetwork(all_oil_value=0.0)) - - assert isinstance(result, pd.Series) - assert result.loc[("AT1", "MWh")] == pytest.approx(0.0) - - def test_no_renewable_oil_production_fossil_equals_total(self): - """Without renewable oil supply, fossil oil should equal total oil demand.""" - result = Final_Energy_by_Carrier__Oil(self._OilNetwork(non_fossil_empty=True)) - - assert isinstance(result, pd.Series) - assert isinstance(result.index, pd.MultiIndex) - assert result.index.names == ["location", "unit"] - assert not result.isna().any() - # 100 (agri) + 100 (res/com) + 300 (transport) = 500 - assert result.loc[("AT1", "MWh")] == pytest.approx(500.0) - - def test_handles_empty_all_oil_without_failing(self): - """Function should work when total oil-withdrawal denominator is empty.""" - result = Final_Energy_by_Carrier__Oil( - self._OilNetwork(all_oil_empty=True, non_fossil_empty=True) - ) - - assert isinstance(result, pd.Series) - assert isinstance(result.index, pd.MultiIndex) - assert result.index.names == ["location", "unit"] - assert not result.isna().any() - assert (result == 0.0).all() - - def test_returns_dataframe_for_aggregate_per_year_false(self): - """Function should return a timeseries DataFrame for aggregate_per_year=False.""" - result = Final_Energy_by_Carrier__Oil( - self._OilNetwork(), - aggregate_per_year=False, - ) - - assert isinstance(result, pd.DataFrame) - assert isinstance(result.index, pd.MultiIndex) - assert result.index.names == ["location", "unit"] - assert isinstance(result.columns, pd.DatetimeIndex) - - # --------------------------------------------------------------------------- # Tests for Final_Energy_by_Carrier__Coal # --------------------------------------------------------------------------- @@ -1823,6 +1545,34 @@ def test_raises_without_energy_totals(self, mock_network: MockPyPSANetwork): with pytest.raises(ValueError): Final_Energy_by_Sector__Transportation(mock_network) + def test_uses_mean_efficiency_and_warns_for_differing_bev_chargers( + self, energy_totals_csv, capsys + ): + """Multiple distinct BEV charger efficiencies fall back to mean with WARNING.""" + network = MockPyPSANetwork( + links=pd.DataFrame( + { + "carrier": ["BEV charger", "BEV charger", "other link"], + "efficiency": [0.8, 0.9, 1.0], + }, + index=["bev_charger_at1", "bev_charger_at2", "other_link_at1"], + ) + ) + + result = Final_Energy_by_Sector__Transportation( + network, energy_totals=energy_totals_csv + ) + + captured = capsys.readouterr() + assert "WARNING: Network includes different efficiencies for BEV chargers" in ( + captured.out + ) + assert isinstance(result, pd.Series) + assert isinstance(result.index, pd.MultiIndex) + assert "location" in result.index.names + assert "unit" in result.index.names + assert len(result) > 0 + # --------------------------------------------------------------------------- # Tests for Final_Energy_by_Sector__Agriculture @@ -1905,8 +1655,9 @@ class TestFinalEnergyBySectorResidentialAndCommercial: class _ResidentialAndCommercialStatisticsAccessor: """Minimal accessor for residential and commercial sector tests.""" - def __init__(self): + def __init__(self, rescom_oil_empty: bool = False): self.calls: list[dict] = [] + self.rescom_oil_empty = rescom_oil_empty @staticmethod def _series_or_frame( @@ -1926,10 +1677,14 @@ def _series_or_frame( "location": location, "unit": unit, } - index_tuples.append(tuple(idx_dict.get(key, f"mock_{key}") for key in groupby)) + index_tuples.append( + tuple(idx_dict.get(key, f"mock_{key}") for key in groupby) + ) index = pd.MultiIndex.from_tuples(index_tuples, names=groupby) - timestamps = pd.date_range("2019-01-01", periods=4, freq="6h", name="snapshot") + timestamps = pd.date_range( + "2019-01-01", periods=4, freq="6h", name="snapshot" + ) if groupby_time: return pd.Series(values, index=index, dtype=float) @@ -1971,19 +1726,32 @@ def energy_balance( if components == "Link" and at_port not in ("bus0", ["bus0"]): return self._series_or_frame(groupby, [], [], "MWh_th", groupby_time) + if bus_carrier == "oil" and self.rescom_oil_empty: + return self._series_or_frame(groupby, [], [], "MWh_th", groupby_time) + if carrier == ["rural biomass boiler", "urban decentral biomass boiler"]: - return self._series_or_frame(groupby, ["AT1", "AT2"], [21.0, 22.0], "MWh_th", groupby_time) + return self._series_or_frame( + groupby, ["AT1", "AT2"], [21.0, 22.0], "MWh_th", groupby_time + ) if bus_carrier == "low voltage": - return self._series_or_frame(groupby, ["AT1", "AT2"], [11.0, 12.0], "MWh_el", groupby_time) + return self._series_or_frame( + groupby, ["AT1", "AT2"], [11.0, 12.0], "MWh_el", groupby_time + ) if bus_carrier == "urban central heat": - return self._series_or_frame(groupby, ["AT1", "AT2"], [13.0, 14.0], "MWh_th", groupby_time) + return self._series_or_frame( + groupby, ["AT1", "AT2"], [13.0, 14.0], "MWh_th", groupby_time + ) if carrier == ["urban decentral gas boiler", "rural gas boiler"]: - return self._series_or_frame(groupby, ["AT1", "AT2"], [15.0, 16.0], "MWh_th", groupby_time) + return self._series_or_frame( + groupby, ["AT1", "AT2"], [15.0, 16.0], "MWh_th", groupby_time + ) - return self._series_or_frame(groupby, ["AT1", "AT2"], [1.0, 2.0], "MWh_th", groupby_time) + return self._series_or_frame( + groupby, ["AT1", "AT2"], [1.0, 2.0], "MWh_th", groupby_time + ) def withdrawal( self, @@ -2008,13 +1776,14 @@ def withdrawal( class _ResidentialAndCommercialNetwork: """Minimal network object exposing a residential/commercial statistics accessor.""" - def __init__(self): - self.statistics = ( - TestFinalEnergyBySectorResidentialAndCommercial._ResidentialAndCommercialStatisticsAccessor() + def __init__(self, rescom_oil_empty: bool = False): + accessor_cls = ( + TestFinalEnergyBySectorResidentialAndCommercial._ResidentialAndCommercialStatisticsAccessor ) + self.statistics = accessor_cls(rescom_oil_empty=rescom_oil_empty) - def _residential_and_commercial_network(self): - return self._ResidentialAndCommercialNetwork() + def _residential_and_commercial_network(self, rescom_oil_empty: bool = False): + return self._ResidentialAndCommercialNetwork(rescom_oil_empty=rescom_oil_empty) def test_returns_series(self): """Test that the function returns a pandas Series.""" @@ -2127,6 +1896,18 @@ def test_returns_dataframe_when_not_aggregated(self): assert not result.empty pd.testing.assert_series_equal(result.sum(axis=1), aggregated_result) + def test_handles_empty_rescom_oil_without_failing(self): + """Function should work when residential/commercial oil demand is empty.""" + network = self._residential_and_commercial_network(rescom_oil_empty=True) + + result = Final_Energy_by_Sector__Residential_and_Commercial(network) + + assert isinstance(result, pd.Series) + assert isinstance(result.index, pd.MultiIndex) + assert result.index.names == ["location", "unit"] + assert not result.isna().any() + assert len(result) > 0 + # --------------------------------------------------------------------------- # Tests for Final_Energy_by_Sector__Industry diff --git a/tests/test_unit_conversion.py b/tests/test_unit_conversion.py index b2a4801..98430aa 100644 --- a/tests/test_unit_conversion.py +++ b/tests/test_unit_conversion.py @@ -114,6 +114,72 @@ def test_raises_keyerror_for_unknown_variable(self, processor: Network_Processor with pytest.raises(KeyError, match="not defined"): processor._get_unit_from_common_definitions("B") + def test_warns_and_uses_first_match_for_multiple_definitions( + self, processor: Network_Processor, capsys + ): + processor.common_dsd = MagicMock() + processor.common_dsd.variable.to_pandas.return_value = pd.DataFrame( + {"variable": ["A", "A"], "unit": ["EJ/yr", "TWh/yr"]} + ) + assert processor._get_unit_from_common_definitions("A") == "EJ/yr" + captured = capsys.readouterr() + assert "WARNING: Multiple definitions found for variable" in captured.out + + def test_raises_runtime_error_when_common_dsd_not_initialized( + self, processor: Network_Processor + ): + processor.common_dsd = None + with pytest.raises(RuntimeError, match="not initialized"): + processor._get_unit_from_common_definitions("A") + + def test_raises_keyerror_when_variable_column_missing( + self, processor: Network_Processor + ): + processor.common_dsd = MagicMock() + processor.common_dsd.variable.to_pandas.return_value = pd.DataFrame( + {"unit": ["EJ/yr"]} + ) + with pytest.raises(KeyError, match="Variable column not found"): + processor._get_unit_from_common_definitions("A") + + def test_raises_keyerror_when_unit_column_missing( + self, processor: Network_Processor + ): + processor.common_dsd = MagicMock() + processor.common_dsd.variable.to_pandas.return_value = pd.DataFrame( + {"variable": ["A"]} + ) + with pytest.raises(KeyError, match="Unit column not found"): + processor._get_unit_from_common_definitions("A") + + def test_parses_first_unit_from_multi_unit_bracket_string( + self, processor: Network_Processor + ): + processor.common_dsd = MagicMock() + processor.common_dsd.variable.to_pandas.return_value = pd.DataFrame( + {"variable": ["A"], "unit": ["['GWh', 'TJ']"]} + ) + assert processor._get_unit_from_common_definitions("A") == "GWh" + + def test_falls_back_to_tj_for_malformed_multi_unit_string( + self, processor: Network_Processor + ): + processor.common_dsd = MagicMock() + processor.common_dsd.variable.to_pandas.return_value = pd.DataFrame( + {"variable": ["A"], "unit": ["[TJ, GWh]"]} + ) + assert processor._get_unit_from_common_definitions("A") == "TJ" + + def test_raises_keyerror_for_nan_unit(self, processor: Network_Processor): + """The NaN check runs before the bracket-parsing check, so a NaN unit + raises the documented KeyError rather than TypeError.""" + processor.common_dsd = MagicMock() + processor.common_dsd.variable.to_pandas.return_value = pd.DataFrame( + {"variable": ["A"], "unit": [float("nan")]} + ) + with pytest.raises(KeyError, match="Unit information not found"): + processor._get_unit_from_common_definitions("A") + class TestConvertUnitsToCommonDefinitions: """Tests for _convert_units_to_common_definitions().""" @@ -157,6 +223,32 @@ def test_converts_variable_to_target_unit(self, processor: Network_Processor): converted = processor._convert_units_to_common_definitions(iam_df) assert converted.unit == ["TWh/yr"] + def test_raises_value_error_when_no_unit_found_for_variable( + self, processor: Network_Processor + ): + processor.common_dsd = MagicMock() + iam_df = MagicMock() + iam_df.variable = ["A"] + var_df = MagicMock() + var_df.unit = [] + iam_df.filter.return_value = var_df + + with pytest.raises(ValueError, match="No unit found for variable"): + processor._convert_units_to_common_definitions(iam_df) + + def test_raises_value_error_when_variable_has_multiple_units( + self, processor: Network_Processor + ): + processor.common_dsd = MagicMock() + iam_df = MagicMock() + iam_df.variable = ["A"] + var_df = MagicMock() + var_df.unit = ["EJ/yr", "TWh/yr"] + iam_df.filter.return_value = var_df + + with pytest.raises(ValueError, match="multiple units"): + processor._convert_units_to_common_definitions(iam_df) + def test_raises_runtime_error_for_missing_variable_definition( self, processor: Network_Processor ): diff --git a/tests/test_workflow.py b/tests/test_workflow.py index 81947a6..5587e60 100644 --- a/tests/test_workflow.py +++ b/tests/test_workflow.py @@ -4,6 +4,7 @@ from pathlib import Path from unittest.mock import MagicMock, patch +import runpy import sys import io @@ -151,6 +152,36 @@ def test_main_execution(self, mock_processor_class): pass +# --------------------------------------------------------------------------- +# Tests for the __main__ execution guard +# --------------------------------------------------------------------------- + + +class TestMainGuard: + """Test the ``if __name__ == \"__main__\"`` module execution guard.""" + + def test_running_module_as_script_invokes_main(self, tmp_path: Path): + """Running workflow.py as __main__ should call main() end-to-end.""" + config_file = tmp_path / "config.yaml" + config_file.write_text("dummy: config\n") + + mock_processor = MagicMock() + mock_processor_class = MagicMock(return_value=mock_processor) + + with patch.object(sys, "argv", ["workflow.py", "--config", str(config_file)]): + with patch( + "pypsa_validation_processing.Network_Processor", mock_processor_class + ): + runpy.run_module( + "pypsa_validation_processing.workflow", run_name="__main__" + ) + + mock_processor_class.assert_called_once_with(config_path=config_file) + mock_processor.read_definitions.assert_called_once() + mock_processor.calculate_variables_values.assert_called_once() + mock_processor.write_output_to_xlsx.assert_called_once() + + # --------------------------------------------------------------------------- # Tests for CLI behavior # ---------------------------------------------------------------------------