Skip to content
Snippets Groups Projects

Test/add xfel calibrate cli tests

Merged Robert Rosca requested to merge test/add-xfel-calibrate-cli-tests into master
1 file
+ 110
7
Compare changes
  • Side-by-side
  • Inline
import ast
import sys
from pathlib import Path
from typing import Generator, List, NamedTuple
from unittest import mock
import pytest
from datetime import date
import nbformat
import pytest
import xfel_calibrate.calibrate as calibrate
import xfel_calibrate.settings as commands
from nbparameterise import extract_parameters
@mock.patch.object(sys, "argv", ["xfel-calibrate", "--help"])
@@ -38,7 +44,17 @@ def test_help_nb(capsys):
assert err == ""
def test_execute_nb(fake_process, capsys, tmpdir):
class CallResult(NamedTuple):
argv: List[str]
out: str
err: str
tmp_path: Path
@pytest.fixture
def call_result(
fake_process, capsys, tmp_path: Path
) -> Generator[CallResult, None, None]:
# Fake calls to slurm
fake_process.register_subprocess(commands.free_nodes_cmd, stdout=["1"])
fake_process.register_subprocess(commands.preempt_nodes_cmd, stdout=["1"])
@@ -51,18 +67,105 @@ def test_execute_nb(fake_process, capsys, tmpdir):
["git", fake_process.any(), "describe", "--tag"], stdout=["0.0.0"]
)
argv = ["xfel-calibrate", "Tutorial", "TEST", "--out-folder", str(tmpdir)]
argv = [
"xfel-calibrate",
"Tutorial",
"TEST",
"--runs",
"1000",
"--out-folder",
str(tmp_path),
]
with mock.patch.object(sys, "argv", argv):
with mock.patch.object(calibrate, "temp_path", str(tmpdir)):
with mock.patch.object(calibrate, "temp_path", str(tmp_path)):
calibrate.run()
out, err = capsys.readouterr()
out, err = capsys.readouterr()
yield CallResult(argv, out, err, tmp_path)
def test_call(call_result: CallResult):
argv, out, err, tmp_path = call_result
assert "sbatch" in out
assert "--job-name xfel_calibrate" in out
assert str(tmpdir) in out
assert str(tmp_path) in out
assert "Submitted job: " in out
assert err == ""
print(out)
@pytest.mark.skip(reason="not implemented")
def test_output_metadata_yml(call_result: CallResult):
# TODO: Finish this test later, not a priority
argv, out, err, tmp_path = call_result
metadata_yml_path = list(tmp_path.glob("**/calibration_metadata.yml")) # noqa
def test_output_ipynb(call_result: CallResult):
argv, out, err, tmp_path = call_result
nb_path = list(tmp_path.glob("**/*/*.ipynb"))
assert len(nb_path) == 1
with nb_path[0].open() as f:
nb = nbformat.read(f, as_version=4)
parameters = {p.name: p.value for p in extract_parameters(nb)}
assert parameters["out_folder"] == str(tmp_path)
assert parameters["sensor_size"] == [10, 30]
assert parameters["random_seed"] == [2345]
assert parameters["runs"] == 1000
def test_output_finalize(call_result: CallResult):
argv, out, err, tmp_path = call_result
finalize_path = list(tmp_path.glob("**/*/finalize.py"))[0]
finalize_ast = ast.parse(finalize_path.read_text())
finalize_kwargs = {
k.arg: ast.literal_eval(k.value)
for k in ast.walk(finalize_ast)
if type(k) is ast.keyword and k.arg != "finaljob"
}
today = date.today()
assert finalize_kwargs["joblist"] == []
assert str(tmp_path) in finalize_kwargs["run_path"]
assert str(tmp_path) in finalize_kwargs["out_path"]
assert finalize_kwargs["project"] == "Tutorial Calculation"
assert finalize_kwargs["calibration"] == "Tutorial Calculation"
assert finalize_kwargs["author"] == "Astrid Muennich"
assert finalize_kwargs["version"] == "0.0.0"
assert str(tmp_path) in finalize_kwargs["report_to"]
assert finalize_kwargs["data_path"] == ""
assert str(today) in finalize_kwargs["request_time"]
assert str(today) in finalize_kwargs["submission_time"]
@pytest.mark.skip(reason="not implemented")
def test_output_rst(call_result: CallResult):
# TODO: Finish this test later, not a priority
argv, out, err, tmp_path = call_result
rst_path = list(tmp_path.glob("**/*/InputParameters.rst"))[0] # noqa
def test_output_sh(call_result: CallResult):
argv, out, err, tmp_path = call_result
sh_path = list(tmp_path.glob("**/*/run_calibrate.sh"))[0]
sh_target = (
"# pycalibration version: 0.0.0\nxfel-calibrate Tutorial TEST "
+ "--runs 1000 --out-folder "
+ str(tmp_path)
)
assert sh_path.read_text() == sh_target
Loading