unzhakov
2788ea0683
- --normalize-to-one renamed to --separate-branches (no alias; the name now describes the output choice: total spectrum with intensity-normalized branches vs separate branches, each normalized to unit area with intensities left to the data user). SpectrumConfig.normalize_to_one -> separate_branches, CSV header word to_one -> separate. - every long option gains a single-letter alias: -N -i -s -m -o -p -L -d -c -V -S -b (only -v/-q existed before) - dry-run drops the top-level Transition: line — the per-branch list below already reports each branch's transition type - version bumped to 0.4.1 (pyproject, __version__, CSV semantics docstring, TODO roadmap entry)
209 lines
7.2 KiB
Python
209 lines
7.2 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_separate_branches_and_tab_separator(
|
|
self, tmp_path: Any
|
|
) -> None:
|
|
src = write_json(tmp_path)
|
|
out = str(tmp_path / "to_one.csv")
|
|
main([src, "--output", out, "--separate-branches", "--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",
|
|
"--separate-branches",
|
|
]
|
|
)
|
|
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_separate_branches(self, tmp_path: Any) -> None:
|
|
out = str(tmp_path / "p32.csv")
|
|
main(["P32", "--output", out, "--separate-branches"])
|
|
|
|
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_paceensdf_co60_dry_run_multi_branch(self, capsys: Any) -> None:
|
|
"""Co60 parses into a healthy multi-branch config (D3): no crash,
|
|
all branch endpoints positive."""
|
|
main(["Co60", "--dry-run"])
|
|
out = capsys.readouterr().out
|
|
assert "Multi-branch mode: 3 branches" in out
|
|
|
|
def test_paceensdf_co60_export(self, tmp_path: Any) -> None:
|
|
out = str(tmp_path / "co60.csv")
|
|
main(["Co60", "--output", out])
|
|
df = read_export(out)
|
|
area = np.trapezoid(df["total_spectrum"], df["energy_MeV"])
|
|
assert np.isclose(area, 1.0, rtol=1e-3)
|
|
eps = header_lines_of(out)
|
|
assert any("E₀=1492.0" in h for h in eps)
|
|
assert not any("excluded" in h for h in eps)
|
|
|
|
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
|