# pylint: disable=missing-class-docstring, missing-function-docstring, no-self-use
"""Tests for the CLI portion of `xfel_calibrate`
These tests cover the CLI interface which is called by the `xfel-calibrate ...`
entrypoint. Some sections of the `calibrate.py` file are still not covered by
the current test cases, this should be improved later on.
"""
import ast
import shlex
import sys
from datetime import date
from pathlib import Path
from unittest import mock
import nbformat
import pytest
from nbparameterise import extract_parameters
import xfel_calibrate.calibrate as calibrate
from tests.test_xfel_calibrate.conftest import (
CalibrateCall,
FakeProcessCalibrate,
MockProposal,
)
class TestBasicCalls:
"""Tests which only call the command line utility `xfel-calibrate` and check
that the expected output is present in stdout
"""
def test_help(self, capsys):
with pytest.raises(SystemExit):
calibrate.run(["xfel-calibrate", "--help"])
out, err = capsys.readouterr()
# Should always be present in these help outputs
assert "positional arguments:" in out
assert "options:" in out
assert err == ""
def test_help_detector(self, capsys):
with pytest.raises(SystemExit):
calibrate.run(["xfel-calibrate", "TEST", "-h"])
out, err = capsys.readouterr()
assert "Notebook for use with the unit and continuous integration" in out
assert "tests." in out
assert err == ""
def test_help_user_notebook(self, capsys):
with pytest.raises(SystemExit):
calibrate.run(["xfel-calibrate", "TEST", "-h"])
out, err = capsys.readouterr()
assert "TEST-USER-NB" in out
assert "/{root}/test-cli.ipynb" in out
assert err == ""
def test_help_bad_config(self):
with pytest.raises(KeyError):
calibrate.run(["xfel-calibrate", "TEST-RAISES-ERRORS", "--help"])
def test_unknown_detector(self, capsys):
with pytest.raises(SystemExit) as exit_exception:
calibrate.run(["xfel-calibrate", "NotADetector", "beep", "-h"])
out, err = capsys.readouterr()
assert exit_exception.value.code == 1
assert "Not one of the known calibrations or detectors" in out
assert err == ""
def test_unknown_detector_h(self, capsys):
with pytest.raises(SystemExit) as exit_exception:
calibrate.run(["xfel-calibrate", "NotADetector", "-h"])
out, err = capsys.readouterr()
assert exit_exception.value.code == 1
assert "Not one of the known detectors" in out
assert err == ""
def test_help_nb(self, capsys):
with pytest.raises(SystemExit):
calibrate.run(["xfel-calibrate", "Tutorial", "TEST", "--help"])
out, err = capsys.readouterr()
# Should always be present in these help outputs
assert "positional arguments:" in out
assert "options:" in out
# Defined in the test notebook, should be propagated to help output
assert "sensor-size" in out
assert "random-seed" in out
assert err == ""
class TestTutorialNotebook:
"""Checks calling `xfel-calibrate` on the `Tutorial TEST` notebook, looks
at the stdout as well as files generated by the call
"""
@pytest.fixture(scope="class", autouse=True)
def fake_process_calibrate(self):
with FakeProcessCalibrate() as fake_process:
yield fake_process
@pytest.fixture(scope="class")
def mock_proposal(self, tmp_path_factory):
return MockProposal(
tmp_path=tmp_path_factory.mktemp("exp"),
instrument="None",
cycle="000000",
proposal="p000000",
runs=0,
sequences=0,
)
@pytest.fixture(scope="function")
def calibrate_call(
self,
mock_proposal: MockProposal,
capsys,
tmp_path,
):
return CalibrateCall(
tmp_path,
capsys,
out_folder=mock_proposal.path_proc,
detector="Tutorial",
cal_type="TEST",
extra_args=["--runs", "1000"],
)
def test_call(
self,
calibrate_call: CalibrateCall,
):
assert "sbatch" in calibrate_call.out
assert "--job-name xfel_calibrate" in calibrate_call.out
assert str(calibrate_call.reports_dir) in calibrate_call.out
assert calibrate_call.err == ""
def test_expected_processes_called(
self,
calibrate_call: CalibrateCall,
fake_process_calibrate: FakeProcessCalibrate,
):
process_calls = [
list(shlex.shlex(p, posix=True, punctuation_chars=True))
if isinstance(p, str)
else p
for p in fake_process_calibrate.calls
]
processes_called = [p[0] for p in process_calls] # List of the process names
assert "sbatch" in processes_called
@pytest.mark.skip(reason="not implemented")
def test_output_metadata_yml(self):
# TODO: Finish this test later, not a priority
# metadata_yml_path = list(self.tmp_path.glob("**/calibration_metadata.yml"))
pass
def test_output_ipynb(self, calibrate_call: CalibrateCall):
notebook_path = calibrate_call.paths.notebooks
assert len(notebook_path) == 1
with notebook_path[0].open() as file:
notebook = nbformat.read(file, as_version=4)
parameters = {p.name: p.value for p in extract_parameters(notebook)}
assert parameters["out_folder"] == str(calibrate_call.out_folder)
assert parameters["sensor_size"] == [10, 30]
assert parameters["random_seed"] == [2345]
assert parameters["runs"] == 1000
def test_output_finalize(
self, mock_proposal: MockProposal, calibrate_call: CalibrateCall
):
# TODO: Specify `feature_version` once we run on python 3.8+
finalize_ast = ast.parse(calibrate_call.paths.finalize.read_text())
today = date.today()
expected_equals = {
"joblist": ["000000"],
"version": "0.0.0",
"data_path": "",
}
expected_contains = {
"request_time": str(today),
"submission_time": str(today),
"cal_work_dir": str(calibrate_call.reports_dir),
# TODO: add a test checking that the out folder is correct
# reffer to: https://git.xfel.eu/gitlab/detectors/pycalibration/issues/52
"out_path": str(mock_proposal.path_proc),
"report_to": str(calibrate_call.reports_dir),
}
# Pull the keyword arguments out of the finalize function call via the AST,
# here we use the keys in `expected_...` to filter which kwargs are parsed
# as some cannot be read
finalize_kwargs = {
k.arg: ast.literal_eval(k.value)
for k in ast.walk(finalize_ast)
if isinstance(k, ast.keyword)
and (k.arg in expected_equals or k.arg in expected_contains)
}
for k, v in expected_equals.items():
assert v == finalize_kwargs[k]
for k, v in expected_contains.items():
assert v in finalize_kwargs[k]
@pytest.mark.skip(reason="not implemented")
def test_output_rst(self, calibrate_call: CalibrateCall):
# TODO: Finish this test later, not a priority
# rst_path = calibrate_call.paths.InputParameters
pass
def test_output_sh(self, calibrate_call: CalibrateCall):
cmd = list(
shlex.shlex(
calibrate_call.paths.run_calibrate.read_text(),
posix=True,
punctuation_chars=True,
)
)
assert (
cmd[0] == "xfel-calibrate"
), f"{calibrate_call.paths.run_calibrate} does not call `xfel-calibrate`"
assert cmd[1:3] == ["Tutorial", "TEST"]
assert {"--out-folder", str(calibrate_call.out_folder)}.issubset(cmd)
assert {"--runs", "1000"}.issubset(cmd)
class TestIntelliList:
@pytest.fixture(scope="class", autouse=True)
def fake_process_calibrate(self):
with FakeProcessCalibrate() as fake_process:
yield fake_process
@pytest.fixture(scope="class")
def mock_proposal(self, tmpdir_factory):
return MockProposal(
tmp_path=Path(tmpdir_factory.mktemp("exp")),
instrument="AGIPD",
cycle="202031",
proposal="p900113",
runs=1,
sequences=1,
)
@pytest.fixture(scope="function")
def calibrate_call(self, mock_proposal: MockProposal, capsys, tmp_path):
return CalibrateCall(
tmp_path,
capsys,
in_folder=mock_proposal.path_raw,
out_folder=mock_proposal.path_proc,
detector="TEST",
cal_type="TEST-CLI",
extra_args=[
"--number",
"10",
"--list-normal",
"1,2,10",
"--list-intellilist",
"1,2,5-8",
"--concurrency-parameter",
"0,1",
],
)
def test_intellilist(self, calibrate_call: CalibrateCall):
assert "--number" in calibrate_call.args
assert "--list-intellilist" in calibrate_call.args
assert "1,2,5-8" in calibrate_call.args
assert len(calibrate_call.paths.notebooks) == 2
for i, notebook_path in enumerate(sorted(calibrate_call.paths.notebooks)):
with notebook_path.open() as file:
notebook = nbformat.read(file, as_version=4)
parameters = {p.name: p.value for p in extract_parameters(notebook)}
assert parameters["number"] == 10
assert parameters["list_normal"] == [1, 2, 10]
assert parameters["list_intellilist"] == [1, 2, 5, 6, 7]
assert parameters["concurrency_parameter"][0] == i
class TestAgipdNotebook:
@pytest.fixture(scope="class", autouse=True)
def fake_process_calibrate(self):
with FakeProcessCalibrate() as fake_process:
yield fake_process
@pytest.fixture(scope="function")
def mock_proposal(self, tmpdir_factory):
return MockProposal(
tmp_path=Path(tmpdir_factory.mktemp("exp")),
instrument="AGIPD",
cycle="202031",
proposal="p900113",
runs=2,
# TODO: update this once extra-data tests can have variable sequences
# sequences=5,
)
@pytest.fixture(scope="function")
def calibrate_call(self, mock_proposal: MockProposal, capsys, tmp_path):
return CalibrateCall(
tmp_path,
capsys,
in_folder=mock_proposal.path_raw,
out_folder=mock_proposal.path_proc / "r0000",
detector="AGIPD",
cal_type="CORRECT",
extra_args=[
"--run",
"0",
"--sequences",
"1-3",
# TODO: enable this when notebook execution tests are ready to be ran
# "--no-cluster-job",
],
)
@pytest.mark.skip(reason="not implemented")
def test_out_folder_correct(self):
# TODO: add a test checking that the out folder is correct
# reffer to: https://git.xfel.eu/gitlab/detectors/pycalibration/issues/52
pass
@pytest.mark.skip(reason="requires extra-data test file sequence options")
def test_files_present(self, calibrate_call: CalibrateCall):
# There should be three notebooks: one pre, then the main, then one post
assert len(calibrate_call.paths.notebooks) == 3
# This is pretty fragile, but the name of notebooks should not change
# (too) often
root_nb_path = calibrate_call.paths.notebooks[0].parent
notebooks = [
root_nb_path / "AGIPD_Correct_and_Verify__sequences__1.ipynb",
root_nb_path / "AGIPD_Correct_and_Verify_Summary_NBC__None__None.ipynb",
root_nb_path / "AGIPD_Retrieve_Constants_Precorrection__None__None.ipynb",
]
assert all(nb in calibrate_call.paths.notebooks for nb in notebooks)
@pytest.mark.skip(reason="not implemented")
def test_nb_sequences(self, calibrate_call: CalibrateCall):
notebook_path = (
calibrate_call.paths.notebooks[0].parent
/ "AGIPD_Correct_and_Verify__sequences__1.ipynb"
)
with notebook_path.open() as file:
notebook = nbformat.read(file, as_version=4)
parameters = {p.name: p.value for p in extract_parameters(notebook)}
# TODO: add test cases for this notebook
print(parameters)
@pytest.mark.skip(reason="not implemented")
def test_nb_summary(self, calibrate_call: CalibrateCall):
notebook_path = (
calibrate_call.paths.notebooks[0].parent
/ "AGIPD_Correct_and_Verify_Summary_NBC__None__None.ipynb"
)
with notebook_path.open() as file:
notebook = nbformat.read(file, as_version=4)
parameters = {p.name: p.value for p in extract_parameters(notebook)}
# TODO: add test cases for this notebook
print(parameters)
@pytest.mark.skip(reason="not implemented")
def test_nb_precorrection(self, calibrate_call: CalibrateCall):
notebook_path = (
calibrate_call.paths.notebooks[0].parent
/ "AGIPD_Retrieve_Constants_Precorrection__None__None.ipynb"
)
with notebook_path.open() as file:
notebook = nbformat.read(file, as_version=4)
# TODO: add test cases for this notebook
parameters = {p.name: p.value for p in extract_parameters(notebook)}
print(parameters)