diff --git a/src/xfel_calibrate/calibrate.py b/src/xfel_calibrate/calibrate.py
index 79ecf612087d5cdc16231d9d1e9e2a375c9f38f9..109cdbd82f3f4df0897b59e25c571031e96f7285 100755
--- a/src/xfel_calibrate/calibrate.py
+++ b/src/xfel_calibrate/calibrate.py
@@ -361,7 +361,7 @@ def make_extended_parser() -> argparse.ArgumentParser:
# the notebook.
for caltype, notebook in det_notebooks.items():
if notebook.get("notebook") is None:
- if notebook.get("user") is None:
+ if notebook.get("user", {}).get("notebook") is None:
raise KeyError(
f"`{detector}` does not have a notebook path, for "
"notebooks that are stored in pycalibration set the "
@@ -419,7 +419,7 @@ def make_extended_parser() -> argparse.ArgumentParser:
)
nb_info["notebook"] = nb_info["user"]["notebook"].format(**vars(user_notebook_args))
- notebook = str(nb_info["notebook"])
+ notebook = nb_info["notebook"]
cvar = nb_info.get("concurrency", {}).get("parameter", None)
diff --git a/tests/test_xfel_calibrate/conftest.py b/tests/test_xfel_calibrate/conftest.py
index c00cfda97c78621930ce5fe2797f94b66ed9068e..61fb9c6f1fb24e61c59bc8fcfe6c66ad321c36ee 100644
--- a/tests/test_xfel_calibrate/conftest.py
+++ b/tests/test_xfel_calibrate/conftest.py
@@ -99,6 +99,11 @@ class MockProposal:
self.path = tmp_path / instrument / cycle / proposal
self.path_raw = self.path / "raw"
self.path_proc = self.path / "proc"
+ self.path_usr = self.path / "usr"
+ self.path_scratch = self.path / "scratch"
+
+ self.path_usr.mkdir(parents=True)
+ self.path_scratch.mkdir()
self.runs = self.create_runs(runs)
diff --git a/tests/test_xfel_calibrate/test_cli.py b/tests/test_xfel_calibrate/test_cli.py
index f0ffeb4af21ededff97abe66e22f669266517e6d..1376f05c52e11d7e8febafc6d24f1ee7cf1ba23c 100644
--- a/tests/test_xfel_calibrate/test_cli.py
+++ b/tests/test_xfel_calibrate/test_cli.py
@@ -8,6 +8,7 @@ the current test cases, this should be improved later on.
import ast
import shlex
+import shutil
import sys
from datetime import date
from pathlib import Path
@@ -55,6 +56,23 @@ class TestBasicCalls:
assert err == ""
+ @mock.patch.object(sys, "argv", ["xfel-calibrate", "TEST", "-h"])
+ def test_help_user_notebook(self, capsys):
+ with pytest.raises(SystemExit):
+ calibrate.run()
+
+ out, err = capsys.readouterr()
+
+ assert "TEST-USER-NB" in out
+ assert "/{root}/test-cli.ipynb" in out
+
+ assert err == ""
+
+ @mock.patch.object(sys, "argv", ["xfel-calibrate", "TEST-RAISES-ERRORS", "--help"])
+ def test_help_bad_config(self):
+ with pytest.raises(KeyError):
+ calibrate.run()
+
@mock.patch.object(sys, "argv", ["xfel-calibrate", "NotADetector", "beep", "-h"])
def test_unknown_detector(self, capsys):
with pytest.raises(SystemExit) as exit_exception:
@@ -307,6 +325,133 @@ class TestIntelliList:
assert parameters["concurrency_parameter"][0] == i
+class TestUserNotebooks:
+ @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: Path):
+ # The user notebook path template for `TEST`/`TEST-USER-NB` is:
+ # `"/{instrument}/{cycle}/p{proposal}/test-cli.ipynb"`
+ #
+ # So by setting `instrument` to tmp_path/TEST we set the user notebook
+ # location to one in our tmp dir
+
+ self.user_nb = mock_proposal.path_usr / "test-cli.ipynb"
+
+ shutil.copy(
+ Path("./notebooks/test/test-cli.ipynb").absolute(), str(self.user_nb)
+ )
+
+ return CalibrateCall(
+ tmp_path,
+ capsys,
+ in_folder=mock_proposal.path_raw,
+ out_folder=mock_proposal.path_proc,
+ detector="TEST",
+ cal_type="TEST-USER-NB",
+ extra_args=[
+ "--root",
+ str(mock_proposal.path_usr),
+ "--number",
+ "10",
+ ],
+ )
+
+ def test_user_notebook_is_test_notebook(self, calibrate_call: CalibrateCall):
+ assert (
+ Path("./notebooks/test/test-cli.ipynb").read_bytes()
+ == self.user_nb.read_bytes()
+ )
+
+ 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["number"] == 10
+
+
+class TestUserVenv:
+ @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: Path):
+ self.user_nb = mock_proposal.path_usr / "test-cli.ipynb"
+
+ shutil.copy(
+ Path("./notebooks/test/test-cli.ipynb").absolute(), str(self.user_nb)
+ )
+
+ return CalibrateCall(
+ tmp_path,
+ capsys,
+ in_folder=mock_proposal.path_raw,
+ out_folder=mock_proposal.path_proc,
+ detector="TEST",
+ cal_type="TEST-USER-NB-VENV",
+ extra_args=[
+ "--root",
+ str(mock_proposal.path_usr),
+ "--number",
+ "10",
+ ],
+ )
+
+ def test_call(self, calibrate_call: CalibrateCall):
+ assert "Running job in user venv at" in calibrate_call.out
+
+ def test_expected_processes_called(
+ self,
+ mock_proposal: MockProposal,
+ 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
+
+ sbatch_cmd = process_calls[processes_called.index("sbatch")]
+ assert any(str(mock_proposal.path_usr / ".venv") in arg for arg in sbatch_cmd)
+
+
class TestAgipdNotebook:
@pytest.fixture(scope="class", autouse=True)
def fake_process_calibrate(self):