dynamisk-rettsstat/tests/test_utvidelser.py

# Copyright (c) 2026 Kjell Tore Guttormsen. MIT License, see LICENSE.
"""Utvidelsene: grensetilfellene gjenskaper kjernen eksakt, og paadrag_tidsserie virker i scenariokjøreren."""

from __future__ import annotations

import importlib
import json
import shutil
import subprocess
import sys
from typing import TYPE_CHECKING

import pytest
import scenario
from conftest import FASIT, MODELL, ROT, kjor

if TYPE_CHECKING:
    from pathlib import Path

UTVIDELSER = sorted(f.stem for f in (MODELL / "utvidelser").glob("*.py") if not f.stem.startswith("_"))


def kjor_fil(data: object, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> tuple[int, str]:
    """Skriv og kjør en scenariofil; returnerer exit-kode og feilmelding."""
    fil = tmp_path / "s.json"
    fil.write_text(json.dumps(data), encoding="utf-8")
    kode = scenario.main([str(fil), "--ut", str(tmp_path / "ut")])
    return kode, capsys.readouterr().err


def test_det_finnes_minst_en_utvidelse() -> None:
    assert "paadrag_tidsserie" in UTVIDELSER


@pytest.mark.parametrize("navn", UTVIDELSER)
def test_grensetilfellene_gjenskaper_kjernen_eksakt(navn: str) -> None:
    modul = scenario.last_utvidelse(navn)
    tilfeller = modul.grensetilfelle()
    assert tilfeller, f"{navn} har ingen grensetilfeller"
    for beskrivelse, avvik in tilfeller:
        assert isinstance(beskrivelse, str)
        assert avvik == 0.0, f"{navn}: {beskrivelse}: {avvik}"


@pytest.mark.parametrize("navn", UTVIDELSER)
def test_grensesnittet(navn: str) -> None:
    modul = scenario.last_utvidelse(navn)
    assert isinstance(modul.NAVN, str)
    assert navn == modul.NAVN
    assert isinstance(modul.BESKRIVELSE, str)
    assert isinstance(modul.PARAMETERE, dict)
    assert callable(modul.simuler)


@pytest.mark.parametrize(("ch0", "fasitfil"), [({}, "kjede_s2"), ({"Z": "inf"}, "p3")])
def test_standardverdiene_gjenskaper_notatets_fil_gjennom_kjoreren(
        ch0: dict[str, object], fasitfil: str, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
    kolonner = ["t", "K", "U", "L", "T", "Q", "tauD"] if fasitfil.startswith("kjede") else ["t", "K", "U", "L", "T"]
    data = {"navn": "u", "modell": "utvidelse:paadrag_tidsserie", "parametere": {"CH0": ch0},
            "utdata": {"format": "notat", "kolonner": kolonner}}
    kode, err = kjor_fil(data, tmp_path, capsys)
    assert kode == 0, err
    assert (tmp_path / "ut" / "u.dat").read_bytes() == (FASIT / "data" / f"{fasitfil}.dat").read_bytes()


def test_tiltak_i_kjeden_virker_som_i_kjernen(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
    data = {"navn": "u", "modell": "utvidelse:paadrag_tidsserie",
            "tiltak": {"andel_utenom": {"andel": 0.2}, "kapasitet": {"faktor": 1.34}},
            "utdata": {"format": "notat", "kolonner": ["t", "K", "U", "L", "T", "Q", "tauD"]}}
    kode, err = kjor_fil(data, tmp_path, capsys)
    assert kode == 0, err
    assert (tmp_path / "ut" / "u.dat").read_bytes() == (FASIT / "data" / "kjede_s5.dat").read_bytes()


def test_faktor_trapp_og_lineaer() -> None:
    modul = importlib.import_module("utvidelser.paadrag_tidsserie")
    punkter = [(12.0, 1.0), (18.0, 1.4), (36.0, 1.4), (48.0, 1.0)]
    assert [modul.faktor(t, "lineaer", punkter) for t in (0.0, 12.0, 15.0, 18.0, 30.0, 42.0, 60.0)] == \
        pytest.approx([1.0, 1.0, 1.2, 1.4, 1.4, 1.2, 1.0])
    trapp = [(12.0, 1.3), (24.0, 1.1)]
    assert [modul.faktor(t, "trapp", trapp) for t in (11.95, 12.0, 23.95, 24.0, 100.0)] == [1.0, 1.3, 1.3, 1.1, 1.1]


def test_bolge_gir_forbigaaende_paadrag(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
    data = {"navn": "u", "modell": "utvidelse:paadrag_tidsserie", "slutt": 120,
            "parametere": {"CH0": {"Z": "inf"}},
            "utvidelse": {"form": "lineaer", "punkter": [[12, 1.0], [18, 1.4], [36, 1.4], [48, 1.0]]},
            "utdata": {"kolonner": ["t", "K", "Omega_faktor"], "hvert": 240}}
    kode, err = kjor_fil(data, tmp_path, capsys)
    assert kode == 0, err
    rader = [linje.split(",") for linje in (tmp_path / "ut" / "u.csv").read_text(encoding="utf-8").splitlines()[1:]]
    assert [r[2] for r in rader] == ["1", "1", "1.4", "1.4", "1", "1", "1", "1", "1", "1", "1"]
    k = [float(r[1]) for r in rader]
    assert max(k) > 1.05
    assert abs(k[-1] - 1.0) < 0.01           # uten flaskehals går K tilbake når presset er borte


@pytest.mark.parametrize(("ekstra", "tekst"), [
    ({"utvidelse": {"punkter": [[12, 1.3], [6, 1.0]]}}, "utvidelse.punkter[1]: månedene må stige"),
    ({"utvidelse": {"punkter": [[12]]}}, "utvidelse.punkter[0]: må være et par"),
    ({"utvidelse": {"punkter": 1.3}}, "utvidelse.punkter: må være en liste"),
    ({"utvidelse": {"form": "glatt"}}, 'utvidelse.form: må være "trapp" eller "lineaer"'),
    ({"utvidelse": {"fart": 2}}, "utvidelse.fart: ukjent felt"),
    ({"styring": {"regulator": "MPC"}}, 'bruker bare "P"'),
    ({"sjokk": {"faktor": 1.2}}, "sjokk: brukes ikke med utvidelser"),
    ({"parametere": {"P": {"shock": 1.5}}}, "parametere.P.shock: brukes ikke av paadrag_tidsserie"),
    ({"parametere": {"CH2": {"c": 2}}}, "parametere.CH2: brukes ikke av modellen"),
    ({"parametere": {"CH0": {"Z": "inf"}}, "tiltak": {"kapasitet": {"faktor": 1.2}}}, "tiltak: virker bare i kjeden"),
])
def test_feil_i_utvidelsens_parametere(ekstra: dict[str, object], tekst: str, tmp_path: Path,
                                       capsys: pytest.CaptureFixture[str]) -> None:
    kode, err = kjor_fil({"modell": "utvidelse:paadrag_tidsserie", **ekstra}, tmp_path, capsys)
    assert kode == 2
    assert tekst in err, err


# ---------------- malen i docs/utvidelser.md ----------------

def malen() -> str:
    """Python-koden mellom markørene i docs/utvidelser.md."""
    tekst = (ROT / "docs" / "utvidelser.md").read_text(encoding="utf-8")
    blokk = tekst.split("<!-- mal:start -->", 1)[1].split("<!-- mal:slutt -->", 1)[0]
    return blokk.split("```python\n", 1)[1].rsplit("```", 1)[0]


@pytest.fixture
def modell_med_mal(tmp_path: Path) -> Path:
    """En kopi av modell/ og fasit/ med malen lagt inn som modell/utvidelser/mal_effektivitet.py."""
    rot = tmp_path / "med_mal"
    for navn in ("modell", "fasit"):
        shutil.copytree(ROT / navn, rot / navn, ignore=shutil.ignore_patterns("__pycache__"))
    (rot / "modell" / "utvidelser" / "mal_effektivitet.py").write_text(malen(), encoding="utf-8")
    return rot


def test_malen_har_eksakt_grensetilfelle(modell_med_mal: Path) -> None:
    modell = modell_med_mal / "modell"
    kode = ("import importlib, json, sys; sys.path[:0] = [sys.argv[1], sys.argv[2]]; "
            "m = importlib.import_module('utvidelser.mal_effektivitet'); print(json.dumps(m.grensetilfelle()))")
    prosess = subprocess.run([sys.executable, "-c", kode, str(modell / "kjerne"), str(modell)],
                             capture_output=True, text=True, check=False)
    assert prosess.returncode == 0, prosess.stderr
    tilfeller = json.loads(prosess.stdout)
    assert len(tilfeller) == 2
    assert all(avvik == 0.0 for _, avvik in tilfeller), tilfeller


def test_malen_kjorer_med_scenariofil(modell_med_mal: Path, tmp_path: Path) -> None:
    fil = tmp_path / "mal.json"
    for faktor, navn in ((1.0, "en"), (1.2, "annen")):
        fil.write_text(json.dumps({"navn": navn, "modell": "utvidelse:mal_effektivitet",
                                   "utvidelse": {"faktor": faktor},
                                   "utdata": {"format": "notat", "kolonner": ["t", "K", "U", "L", "T"]}}),
                       encoding="utf-8")
        prosess = kjor("--modell", str(modell_med_mal), "scenario", str(fil), "--ut", str(tmp_path / navn))
        assert prosess.returncode == 0, prosess.stderr
    assert (tmp_path / "en" / "en.dat").read_bytes() == (FASIT / "data" / "p3.dat").read_bytes()
    assert (tmp_path / "annen" / "annen.dat").read_bytes() != (FASIT / "data" / "p3.dat").read_bytes()


# ---------------- uavhengig av utvidelsens egen beregning: gjennom kjøreren mot fasit ----------------

NOTATFILER = [(navn, fil) for navn in UTVIDELSER
              for fil in getattr(scenario.last_utvidelse(navn), "GJENSKAPER_NOTATET", ())]


@pytest.mark.parametrize("navn", UTVIDELSER)
def test_utvidelsen_oppgir_notatfiler(navn: str) -> None:
    filer = getattr(scenario.last_utvidelse(navn), "GJENSKAPER_NOTATET", None)
    if filer is None:
        pytest.skip(f"{navn} oppgir ikke GJENSKAPER_NOTATET; grensetilfellet er bare sjekket med egen beregning")
    assert filer, f"{navn}: GJENSKAPER_NOTATET er tom"
    for fil in filer:
        assert (ROT / "eksempler" / "scenarioer" / "notatet" / f"{fil}.json").is_file(), fil
        assert (FASIT / "data" / f"{fil}.dat").is_file(), fil


@pytest.mark.parametrize(("navn", "fil"), NOTATFILER)
def test_standardverdiene_gjenskaper_notatfilene_gjennom_kjoreren(
        navn: str, fil: str, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
    data = json.loads((ROT / "eksempler" / "scenarioer" / "notatet" / f"{fil}.json").read_text(encoding="utf-8"))
    data["modell"] = f"utvidelse:{navn}"
    kode, err = kjor_fil(data, tmp_path, capsys)
    assert kode == 0, err
    assert (tmp_path / "ut" / f"{fil}.dat").read_bytes() == (FASIT / "data" / f"{fil}.dat").read_bytes()


def test_utvidelsen_stoppes_ved_ubalanse_og_kan_kalibreres(tmp_path: Path,
                                                          capsys: pytest.CaptureFixture[str]) -> None:
    data: dict[str, object] = {"navn": "u", "modell": "utvidelse:paadrag_tidsserie",
                               "parametere": {"P": {"gamma": 0.15}}}
    kode, err = kjor_fil(data, tmp_path, capsys)
    assert kode == 2
    assert "startpunktet er ikke en likevekt" in err
    data["startlikevekt"] = "kalibrer"
    kode, err = kjor_fil(data, tmp_path, capsys)
    assert kode == 0, err