unzhakov
9c3dd80b5c
Remove the duplicated inline paceENSDF config builder in the CLI. It renormalized branch intensities before filtering, diverging from the library decay_info_to_config path that library callers use (audit finding D5). CLI paceENSDF configs now come from create_config_from_source exactly once per run. Regression test spies on create_config_from_source during a CLI run.
192 lines
6.4 KiB
Python
192 lines
6.4 KiB
Python
"""CLI flow tests: command-line options must reach the config in every
|
|
data-source flow (JSON file and paceENSDF), and degenerate exports must
|
|
exit non-zero without writing a file (regression D2/D4).
|
|
|
|
The CLI is invoked in-process via beta_spectrum.cli.main(argv) so the
|
|
paceENSDF dataset cache (loaded once per process) is shared with the
|
|
rest of the test session.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from beta_spectrum.cli import main
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _base_json() -> Dict[str, Any]:
|
|
return {
|
|
"parent_symbol": "Tc99",
|
|
"Z_parent": 43,
|
|
"Z_daughter": 44,
|
|
"A_number": 99,
|
|
"endpoint_MeV": 0.294,
|
|
"transition_type": "A",
|
|
}
|
|
|
|
|
|
def write_json(tmp_path: Any, extra: Optional[Dict[str, Any]] = None) -> str:
|
|
data = _base_json()
|
|
if extra:
|
|
data.update(extra)
|
|
path = tmp_path / "input.json"
|
|
path.write_text(json.dumps(data))
|
|
return str(path)
|
|
|
|
|
|
def header_lines_of(path: str) -> List[str]:
|
|
lines: List[str] = []
|
|
with open(path, "r") as f:
|
|
for line in f:
|
|
if line.startswith("#"):
|
|
lines.append(line.rstrip("\n"))
|
|
else:
|
|
break
|
|
return lines
|
|
|
|
|
|
def read_export(path: str) -> pd.DataFrame:
|
|
return pd.read_csv(path, sep=None, engine="python", comment="#")
|
|
|
|
|
|
def n_points_of(header: List[str]) -> int:
|
|
for h in header:
|
|
if h.startswith("# n_points:"):
|
|
return int(h.split(":")[1])
|
|
raise AssertionError(f"n_points header missing: {header}")
|
|
|
|
|
|
class TestCLIJsonFlows:
|
|
"""Options must propagate through the JSON config flow (D2)."""
|
|
|
|
def test_json_honors_normalize_to_one_and_tab_separator(
|
|
self, tmp_path: Any
|
|
) -> None:
|
|
src = write_json(tmp_path)
|
|
out = str(tmp_path / "to_one.csv")
|
|
main([src, "--output", out, "--normalize-to-one", "--csv-separator", "tab"])
|
|
|
|
with open(out) as f:
|
|
body = "".join(ln for ln in f if not ln.startswith("#"))
|
|
assert "\t" in body, "CSV must be tab-separated"
|
|
|
|
df = read_export(out)
|
|
assert "total_spectrum" not in df.columns
|
|
area = np.trapezoid(df["branch_1_spectrum"], df["energy_MeV"])
|
|
assert np.isclose(area, 1.0, rtol=1e-3)
|
|
|
|
def test_json_defaults_keep_total_column(self, tmp_path: Any) -> None:
|
|
src = write_json(tmp_path)
|
|
out = str(tmp_path / "default.csv")
|
|
main([src, "--output", out])
|
|
|
|
df = read_export(out)
|
|
assert "total_spectrum" in df.columns
|
|
area = np.trapezoid(df["total_spectrum"], df["energy_MeV"])
|
|
assert np.isclose(area, 1.0, rtol=1e-3)
|
|
|
|
def test_json_cli_overrides_file_physics_values(self, tmp_path: Any) -> None:
|
|
src = write_json(tmp_path, {"e_step_MeV": 0.001, "neutrino_mass_MeV": 0.0})
|
|
out = str(tmp_path / "overrides.csv")
|
|
main([src, "--output", out, "--e-step", "0.02", "--neutrino-mass", "0.15"])
|
|
|
|
header = header_lines_of(out)
|
|
assert any("neutrino_mass_MeV: 0.150000" in h for h in header)
|
|
expected_points = int(np.arange(0.02, 0.294, 0.02).size)
|
|
assert n_points_of(header) == expected_points
|
|
|
|
def test_json_file_values_used_when_flags_absent(self, tmp_path: Any) -> None:
|
|
src = write_json(tmp_path, {"e_step_MeV": 0.005})
|
|
out = str(tmp_path / "file_values.csv")
|
|
main([src, "--output", out])
|
|
|
|
header = header_lines_of(out)
|
|
expected_points = int(np.arange(0.005, 0.294, 0.005).size)
|
|
assert n_points_of(header) == expected_points
|
|
|
|
|
|
class TestCLIDegenerateFlows:
|
|
"""Degenerate exports (m_nu above every endpoint) must fail cleanly."""
|
|
|
|
def test_degenerate_intensity_mode_exits_nonzero_no_file(
|
|
self, tmp_path: Any
|
|
) -> None:
|
|
src = write_json(tmp_path)
|
|
out = str(tmp_path / "deg.csv")
|
|
with pytest.raises(SystemExit) as exc:
|
|
main([src, "--output", out, "--neutrino-mass", "0.35"])
|
|
assert exc.value.code == 1
|
|
assert not (tmp_path / "deg.csv").exists()
|
|
|
|
def test_degenerate_to_one_mode_exits_nonzero_no_file(self, tmp_path: Any) -> None:
|
|
src = write_json(tmp_path)
|
|
out = str(tmp_path / "deg_to_one.csv")
|
|
with pytest.raises(SystemExit) as exc:
|
|
main(
|
|
[
|
|
src,
|
|
"--output",
|
|
out,
|
|
"--neutrino-mass",
|
|
"0.35",
|
|
"--normalize-to-one",
|
|
]
|
|
)
|
|
assert exc.value.code == 1
|
|
assert not (tmp_path / "deg_to_one.csv").exists()
|
|
|
|
|
|
@pytest.mark.requires_ensdf
|
|
class TestCLIPaceENSDF:
|
|
"""paceENSDF flow honors the same export options."""
|
|
|
|
def test_paceensdf_honors_normalize_to_one(self, tmp_path: Any) -> None:
|
|
out = str(tmp_path / "p32.csv")
|
|
main(["P32", "--output", out, "--normalize-to-one"])
|
|
|
|
df = read_export(out)
|
|
assert "total_spectrum" not in df.columns
|
|
area = np.trapezoid(df["branch_1_spectrum"], df["energy_MeV"])
|
|
assert np.isclose(area, 1.0, rtol=1e-3)
|
|
|
|
def test_paceensdf_degenerate_exits_nonzero_no_file(self, tmp_path: Any) -> None:
|
|
out = str(tmp_path / "tc99deg.csv")
|
|
with pytest.raises(SystemExit) as exc:
|
|
main(["Tc99", "--output", out, "--neutrino-mass", "0.35"])
|
|
assert exc.value.code == 1
|
|
assert not (tmp_path / "tc99deg.csv").exists()
|
|
|
|
def test_cli_builds_paceensdf_config_via_library(
|
|
self, monkeypatch: Any, capsys: Any
|
|
) -> None:
|
|
"""CLI must delegate paceENSDF config building to the library
|
|
(create_config_from_source), not duplicate it inline (D5)."""
|
|
import beta_spectrum.nuclear_data as nuclear_data
|
|
|
|
real_builder = nuclear_data.create_config_from_source
|
|
calls: List[Dict[str, Any]] = []
|
|
|
|
def spy(source: str, **kwargs: Any) -> Any:
|
|
calls.append({"source": source, **kwargs})
|
|
return real_builder(source, **kwargs)
|
|
|
|
monkeypatch.setattr(nuclear_data, "create_config_from_source", spy)
|
|
|
|
main(["Cs137", "--dry-run", "--intensity-cutoff", "0.9"])
|
|
capsys.readouterr()
|
|
|
|
assert len(calls) == 1, (
|
|
"CLI must delegate paceENSDF config building to the library "
|
|
"builder exactly once"
|
|
)
|
|
assert calls[0]["source"] == "paceENSDF"
|
|
assert calls[0]["nuclide"] == "Cs137"
|
|
assert calls[0]["intensity_cutoff"] == 0.9
|