Source code for message_ix_models.tests.project.test_ssp

from typing import TYPE_CHECKING

import pytest
from genno import ComputationError, Computer

from message_ix_models.project.ssp import (
    SSP,
    SSP_2017,
    SSP_2024,
    generate,
    parse,
    ssp_field,
)
from message_ix_models.project.ssp.data import SSPOriginal, SSPUpdate

if TYPE_CHECKING:
    from message_ix_models import Context


[docs] def test_generate(tmp_path, test_context): # Function runs generate(test_context, base_dir=tmp_path) # Two XML files are created assert 2 == len(list(tmp_path.glob("*.xml")))
[docs] def test_enum(): # Enumerations have the expected length assert 5 == len(SSP_2017) assert 5 == len(SSP_2024) # Members can be accessed by ID a = SSP_2017["1"] # …or by value b = SSP_2017(1) # …all retrieving the same member assert a == b # __getattr__ lookup does not invoke _missing_ with pytest.raises(KeyError): SSP_2017[1] # Same SSP ID from different enums are not equivalent assert SSP_2017["1"] != SSP_2024["1"] assert SSP_2017["1"] is not SSP_2024["1"] # NB Ignored because of https://github.com/python/mypy/issues/7568 assert SSP["1"] != SSP_2024["1"] # type: ignore [misc]
[docs] @pytest.mark.parametrize( "expected, value", ( (SSP_2017["1"], SSP_2017["1"]), # Literal value (SSP_2017["1"], "1"), # String ID as appears in the codelist (SSP_2017["1"], "SSP1"), # Prefixed by "SSP" (SSP_2017["1"], 1), # Integer ), ) def test_parse(value, expected): assert expected == parse(value)
[docs] def test_ssp_field() -> None: from dataclasses import dataclass @dataclass class Foo: bar: ssp_field = ssp_field(default=SSP_2017["1"]) baz: ssp_field = ssp_field(default=SSP_2024["5"]) # Can be instantiated with no arguments f = Foo() assert SSP_2017["1"] is f.bar assert SSP_2024["5"] is f.baz # Can be instantiated with different values f = Foo(bar=SSP_2017["3"], baz=SSP_2024["3"]) assert SSP_2017["3"] is f.bar assert SSP_2024["3"] is f.baz # Values can be set and are passed through parse() f.bar = "SSP2" f.baz = "SSP4" assert SSP_2017["2"] is f.bar assert SSP_2017["4"] is f.baz
[docs] def test_cli(mix_models_cli): mix_models_cli.assert_exit_0(["ssp", "gen-structures", "--dry-run"])
[docs] class TestSSPOriginal: @pytest.mark.usefixtures("ssp_test_data") @pytest.mark.parametrize( "source", ( "ICONICS:SSP(2017).1", "ICONICS:SSP(2017).2", "ICONICS:SSP(2017).3", "ICONICS:SSP(2017).4", "ICONICS:SSP(2017).5", ), ) @pytest.mark.parametrize( "source_kw", ( dict(measure="POP", model="OECD Env-Growth"), dict(measure="GDP", model="OECD Env-Growth"), # Excess keyword arguments pytest.param( dict(measure="GDP", model="OECD Env-Growth", foo="bar"), marks=pytest.mark.xfail(raises=TypeError), ), ), ) def test_add_tasks(self, test_context: "Context", source, source_kw: dict) -> None: # FIXME The following should be redundant, but appears mutable on GHA linux and # Windows runners. test_context.model.regions = "R14" c = Computer() keys = SSPOriginal.add_tasks( c, context=test_context, source=source, **source_kw ) # Preparation of data runs successfully result = c.get(keys[0]) # Data has the expected dimensions assert ("n", "y") == result.dims # Data is complete assert 14 == len(result.coords["n"]) assert 14 == len(result.coords["y"])
[docs] class TestSSPUpdate: @pytest.mark.usefixtures( "ssp_test_data", # For release=preview "ssp_user_data", # For other values of `release` ) @pytest.mark.parametrize( "source", ( "ICONICS:SSP(2024).1", "ICONICS:SSP(2024).2", "ICONICS:SSP(2024).3", "ICONICS:SSP(2024).4", "ICONICS:SSP(2024).5", ), ) @pytest.mark.parametrize( "release, measure, model", ( ("preview", "GDP", "OECD ENV-Growth 2023"), ("preview", "GDP", "IIASA GDP 2023"), ("preview", "POP", ""), ("3.0", "GDP", "OECD ENV-Growth 2023"), ("3.0", "GDP", "IIASA GDP 2023"), ("3.0", "POP", ""), ("3.0.1", "GDP", "OECD ENV-Growth 2023"), ("3.0.1", "GDP", "IIASA GDP 2023"), ("3.0.1", "POP", ""), ("3.1", "GDP", "OECD ENV-Growth 2023"), ("3.1", "GDP", "IIASA GDP 2023"), ("3.1", "POP", ""), ("3.2.beta", "GDP", "OECD ENV-Growth 2025"), pytest.param( "3.2.beta", "GDP", "IIASA GDP 2025", marks=pytest.mark.xfail(raises=ComputationError, reason="No data"), ), ("3.2.beta", "POP", ""), ), ) def test_add_tasks( self, test_context: "Context", source: str, release: str, measure: str, model: str, ) -> None: # FIXME The following should be redundant, but appears mutable on GHA linux and # Windows runners. test_context.model.regions = "R14" # Prepare source_kw source_kw = dict(release=release, measure=measure) if model: source_kw.update(model=model) if release in ("3.2.beta",) and measure == "GDP": # Disambiguate units for this release source_kw["unit"] = "billion USD_2017/yr" c = Computer() keys = SSPUpdate.add_tasks( c, context=test_context, strict=True, source=source, **source_kw ) # Preparation of data runs successfully result = c.get(keys[0]) # Data has the expected dimensions assert ("n", "y") == result.dims # Data is complete assert 14 == len(result.coords["n"]) assert 14 == len(result.coords["y"]) if release == "preview": return # Fuzzed/random data, not meaningful # Check for apparent double-counting: if in 2025, values will be at least twice # the 2020 values; if in 2020, roughly the opposite ratio = (result.sel(y=2025) / result.sel(y=2020)).to_series() check = (ratio < 0.55) | (1.9 < ratio) assert not check.any(), f"Possible double-counting:\n{ratio[check].to_string()}"